-
-
+
+
+
Worker WebSocket
+ ab-av1
+
+
+
+
+
URL
+
+ {@worker_socket_url}?token=<worker-token>
+
+
+
+
+
Token
+
+ <%= case @worker_token_state do %>
+ <% {:ok, token, _fingerprint} -> %>
+ {token}
+ <% :error -> %>
+ not configured
+ <% end %>
+
+
+
+
+
Token fingerprint
+
+ <%= case @worker_token_state do %>
+ <% {:ok, _token, fingerprint} -> %>
+ {fingerprint}
+ <% :error -> %>
+ not configured
+ <% end %>
-
- Syncing {@service_type}... {@sync_progress}%
+
+
+
Worker sessions
+ <.link
+ navigate={~p"/workers"}
+ class="rounded-full border border-cyan-900 bg-cyan-950 px-3 py-1 text-xs text-cyan-300 hover:border-cyan-700 hover:text-white"
+ >
+ Open workers
+
- <% end %>
+
"""
end
@@ -1048,6 +814,21 @@ defmodule ReencodarrWeb.DashboardLive do
defp format_number(_), do: "—"
+ defp worker_token_state do
+ case Application.get_env(:reencodarr, :worker_token) do
+ token when is_binary(token) ->
+ digest =
+ :crypto.hash(:sha256, token)
+ |> Base.encode16(case: :lower)
+ |> String.slice(0, 12)
+
+ {:ok, token, "sha256:#{digest}"}
+
+ _ ->
+ :error
+ end
+ end
+
defp format_completed(stats) do
total = stats.total_videos || 0
encoded = stats.encoded || 0
@@ -1168,7 +949,6 @@ defmodule ReencodarrWeb.DashboardLive do
video={@crf_search_video}
results={@crf_search_results}
sample={@crf_search_sample}
- progress={@crf_progress}
queue_count={@queue_counts.crf_searcher}
queue_items={@queue_items.crf_searcher}
status={@service_status.crf_searcher}
@@ -1224,6 +1004,8 @@ defmodule ReencodarrWeb.DashboardLive do
syncing={@syncing}
sync_progress={@sync_progress}
service_type={@service_type}
+ worker_token_state={@worker_token_state}
+ worker_socket_url={@worker_socket_url}
/>
@@ -1385,6 +1167,13 @@ defmodule ReencodarrWeb.DashboardLive do
Process.send_after(self(), :update_dashboard_data, 5_000)
end
+ defp worker_socket_url do
+ ReencodarrWeb.Endpoint.url()
+ |> String.replace_prefix("https://", "wss://")
+ |> String.replace_prefix("http://", "ws://")
+ |> Kernel.<>("/workers/socket/websocket")
+ end
+
defp stats_display(nil) do
%{
total_videos: "—",
diff --git a/lib/reencodarr_web/live/workers_live.ex b/lib/reencodarr_web/live/workers_live.ex
new file mode 100644
index 00000000..ad337a61
--- /dev/null
+++ b/lib/reencodarr_web/live/workers_live.ex
@@ -0,0 +1,409 @@
+defmodule ReencodarrWeb.WorkersLive do
+ @moduledoc """
+ LiveView for connected ab-av1 workers and their current status.
+ """
+
+ use ReencodarrWeb, :live_view
+
+ alias Reencodarr.AbAv1.WorkerSessions
+ alias Reencodarr.Dashboard.Events
+ alias Reencodarr.Formatters
+ alias Reencodarr.Media
+ alias Reencodarr.Rules
+
+ import ReencodarrWeb.CrfSearchComponents
+
+ @refresh_interval 5_000
+
+ @impl true
+ def mount(_params, _session, socket) do
+ socket = assign(socket, :workers, WorkerSessions.list())
+
+ if connected?(socket) do
+ Phoenix.PubSub.subscribe(Reencodarr.PubSub, Events.channel())
+ Process.send_after(self(), :refresh_workers, @refresh_interval)
+ end
+
+ {:ok, socket}
+ end
+
+ @impl true
+ def handle_info(:refresh_workers, socket) do
+ Process.send_after(self(), :refresh_workers, @refresh_interval)
+ {:noreply, assign(socket, :workers, WorkerSessions.list())}
+ end
+
+ @impl true
+ def handle_info({:worker_sessions_updated, %{sessions: sessions}}, socket) do
+ {:noreply, assign(socket, :workers, sessions)}
+ end
+
+ def handle_info({_event, _data}, socket), do: {:noreply, socket}
+
+ @impl true
+ def handle_event("pause_worker_crf_search", %{"worker-id" => worker_id}, socket) do
+ control_worker(socket, worker_id, :pause, "Worker pause requested")
+ end
+
+ def handle_event("resume_worker_crf_search", %{"worker-id" => worker_id}, socket) do
+ control_worker(socket, worker_id, :resume, "Worker resume requested")
+ end
+
+ def handle_event("stop_worker_crf_search", %{"worker-id" => worker_id}, socket) do
+ control_worker(socket, worker_id, :stop, "Worker stop requested")
+ end
+
+ @impl true
+ def render(assigns) do
+ ~H"""
+
+
+
+
+
ab-av1
+
Workers
+
+ Connected worker sessions and their current job state.
+
+
+
+
+
+ Connected {length(@workers)}
+
+ <.link
+ navigate={~p"/"}
+ class="rounded-full border border-gray-800 bg-gray-900 px-3 py-1 text-xs text-gray-300 hover:border-cyan-700 hover:text-white"
+ >
+ Dashboard
+
+
+
+
+ <%= if @workers == [] do %>
+
+ No workers connected.
+
+ <% else %>
+
+ <%= for worker <- @workers do %>
+
+
+
+
+
{worker.client_worker_id}
+
+ server: {worker.server_worker_id}
+
+
+
+ {worker_status(worker)}
+
+
+
+ Resources
+
+
+
CPU {worker_cpu(worker)}
+
Mem {worker_memory(worker)}
+
Disk {worker_disk(worker)}
+
+
+
+
+ <%= case worker_phase(worker) do %>
+ <% :receiving_input -> %>
+ <.transfer_panel
+ worker={worker}
+ video={active_video(worker)}
+ title="Receiving Input"
+ />
+ <% :input_ready -> %>
+ <.transfer_panel
+ worker={worker}
+ video={active_video(worker)}
+ title="Input Ready"
+ />
+ <% :crf_searching -> %>
+ <.crf_search_panel
+ video={worker_crf_video(worker)}
+ results={worker_crf_results(worker)}
+ sample={worker_crf_sample(worker)}
+ progress={worker.crf_search_progress || :none}
+ status={worker_crf_status(worker)}
+ show_controls={true}
+ show_queue={false}
+ show_empty_chart={true}
+ suspend_event="pause_worker_crf_search"
+ resume_event="resume_worker_crf_search"
+ fail_event="stop_worker_crf_search"
+ worker_id={worker.server_worker_id}
+ />
+ <% :idle -> %>
+ <.idle_panel />
+ <% end %>
+
+
+ <% end %>
+
+ <% end %>
+
+
+ """
+ end
+
+ defp idle_panel(assigns) do
+ ~H"""
+
+
Idle
+
Waiting for work.
+
+ """
+ end
+
+ defp control_worker(socket, worker_id, action, message) do
+ Phoenix.PubSub.broadcast(
+ Reencodarr.PubSub,
+ ReencodarrWeb.WorkerChannel.worker_control_topic(worker_id),
+ {:worker_control, action}
+ )
+
+ {:noreply, put_flash(socket, :info, message)}
+ end
+
+ defp transfer_panel(assigns) do
+ ~H"""
+
+
+
{@title}
+
+ {format_number(@worker.transfer_progress && @worker.transfer_progress.percent)}%
+
+
+
+ <%= if @video do %>
+
+
{Path.basename(@video.path)}
+
+ {Formatters.file_size(@video.size)}
+ {format_dimensions(@video)}
+ <%= if @video.hdr do %>
+ HDR
+ <% end %>
+ Target: {Rules.vmaf_target(@video)} VMAF
+
+
+ <% end %>
+
+ <%= if @worker.transfer_progress do %>
+
+
+
+
+ {format_transfer_bytes(@worker.transfer_progress)}
+ <%= if chunk_progress = transfer_chunk_progress(@worker.transfer_progress) do %>
+ Chunk {chunk_progress}
+ <% end %>
+ {format_throughput(Map.get(@worker.transfer_progress, :bytes_per_second))}
+ ETA {format_eta(Map.get(@worker.transfer_progress, :eta))}
+
+
+ <% else %>
+
Waiting for work.
+ <% end %>
+
+ """
+ end
+
+ defp worker_status(worker) do
+ case worker_phase(worker) do
+ :idle -> "Idle"
+ :receiving_input -> "Receiving input"
+ :input_ready -> "Input ready"
+ :crf_searching -> "CRF search"
+ end
+ end
+
+ defp worker_crf_status(%{phase: :crf_searching}), do: :processing
+
+ defp worker_crf_status(_worker), do: :idle
+
+ defp worker_phase(%{phase: phase})
+ when phase in [:idle, :receiving_input, :input_ready, :crf_searching],
+ do: phase
+
+ defp worker_phase(%{active_video_id: nil}), do: :idle
+
+ defp worker_phase(%{transfer_progress: progress}) when not is_nil(progress),
+ do: :receiving_input
+
+ defp worker_phase(%{active_video_id: video_id}) when is_integer(video_id), do: :crf_searching
+ defp worker_phase(_worker), do: :idle
+
+ defp worker_crf_video(worker) do
+ case active_video(worker) do
+ %Media.Video{} = video ->
+ %{
+ video_id: video.id,
+ filename: Path.basename(video.path),
+ video_size: video.size,
+ width: video.width,
+ height: video.height,
+ hdr: video.hdr,
+ target_vmaf: Rules.vmaf_target(video)
+ }
+
+ nil ->
+ nil
+ end
+ end
+
+ defp worker_crf_results(worker) do
+ case active_video_id(worker) do
+ nil ->
+ []
+
+ video_id ->
+ video_id
+ |> Media.get_vmafs_for_video()
+ |> Enum.sort_by(& &1.crf)
+ |> Enum.map(fn vmaf ->
+ %{crf: vmaf.crf, score: vmaf.score, percent: vmaf.percent}
+ end)
+ end
+ end
+
+ defp worker_crf_sample(%{
+ crf_search_progress: %{crf: crf, sample_num: sample_num, total_samples: total_samples}
+ })
+ when is_number(crf) and is_integer(sample_num) and is_integer(total_samples) do
+ %{crf: crf, sample_num: sample_num, total_samples: total_samples}
+ end
+
+ defp worker_crf_sample(_worker), do: nil
+
+ defp active_video(worker) do
+ case active_video_id(worker) do
+ nil -> nil
+ video_id -> Media.get_video(video_id)
+ end
+ end
+
+ defp active_video_id(%{active_video_id: video_id}) when is_integer(video_id), do: video_id
+ defp active_video_id(%{crf_search_progress: %{video_id: video_id}}), do: video_id
+ defp active_video_id(%{transfer_progress: %{video_id: video_id}}), do: video_id
+ defp active_video_id(_worker), do: nil
+
+ defp worker_cpu(%{resource_usage: %{cpu_percent: cpu_percent}}) when is_number(cpu_percent),
+ do: "#{format_number(cpu_percent)}%"
+
+ defp worker_cpu(_worker), do: "-"
+
+ defp worker_memory(%{resource_usage: %{memory_bytes: memory_bytes} = usage})
+ when is_integer(memory_bytes) do
+ case Map.get(usage, :memory_total_bytes) do
+ total when is_integer(total) and total > 0 ->
+ "#{Formatters.file_size(memory_bytes)} / #{Formatters.file_size(total)}"
+
+ _ ->
+ Formatters.file_size(memory_bytes)
+ end
+ end
+
+ defp worker_memory(_worker), do: "-"
+
+ defp worker_disk(%{resource_usage: %{disk_free_bytes: disk_free_bytes} = usage})
+ when is_integer(disk_free_bytes) do
+ case Map.get(usage, :disk_total_bytes) do
+ total when is_integer(total) and total > 0 ->
+ "#{Formatters.file_size(disk_free_bytes)} free / #{Formatters.file_size(total)}"
+
+ _ ->
+ "#{Formatters.file_size(disk_free_bytes)} free"
+ end
+ end
+
+ defp worker_disk(_worker), do: "-"
+
+ defp worker_resource_usage?(%{resource_usage: usage}) when is_map(usage),
+ do: map_size(usage) > 0
+
+ defp worker_resource_usage?(_worker), do: false
+
+ defp format_number(number) when is_integer(number), do: Integer.to_string(number)
+
+ defp format_number(number) when is_float(number) do
+ :erlang.float_to_binary(number, decimals: 1)
+ end
+
+ defp format_number(nil), do: "-"
+
+ defp progress_width(percent) when is_number(percent) do
+ percent
+ |> max(0)
+ |> min(100)
+ |> format_number()
+ |> Kernel.<>("%")
+ end
+
+ defp progress_width(_percent), do: "0%"
+
+ defp format_dimensions(%{width: width, height: height})
+ when is_integer(width) and is_integer(height),
+ do: "#{width}x#{height}"
+
+ defp format_dimensions(_video), do: "unknown resolution"
+
+ defp format_throughput(bytes_per_second)
+ when is_integer(bytes_per_second) and bytes_per_second >= 0 do
+ "#{Formatters.file_size(bytes_per_second)}/s"
+ end
+
+ defp format_throughput(_), do: "-"
+
+ defp format_eta(seconds) when is_integer(seconds) and seconds >= 0, do: "#{seconds}s"
+ defp format_eta(_), do: "-"
+
+ defp format_transfer_bytes(bytes_sent, total_bytes)
+ when is_integer(bytes_sent) and bytes_sent >= 0 and is_integer(total_bytes) and
+ total_bytes > 0 do
+ "#{Formatters.file_size(bytes_sent)} / #{Formatters.file_size(total_bytes)}"
+ end
+
+ defp format_transfer_bytes(bytes_sent, _total_bytes)
+ when is_integer(bytes_sent) and bytes_sent >= 0,
+ do: Formatters.file_size(bytes_sent)
+
+ defp format_transfer_bytes(_bytes_sent, _total_bytes), do: "-"
+
+ defp format_transfer_bytes(%{bytes_sent: bytes_sent, total_bytes: total_bytes}),
+ do: format_transfer_bytes(bytes_sent, total_bytes)
+
+ defp transfer_chunk_progress(%{chunk_index: chunk_index, total_chunks: total_chunks})
+ when is_integer(chunk_index) and is_integer(total_chunks) and total_chunks > 0 do
+ "#{chunk_index + 1} / #{total_chunks}"
+ end
+
+ defp transfer_chunk_progress(_progress), do: nil
+
+ defp status_badge_class(worker) do
+ case worker_phase(worker) do
+ :idle ->
+ "inline-flex rounded-full border border-emerald-900 bg-emerald-950 px-2 py-1 text-xs font-medium text-emerald-300"
+
+ _active ->
+ "inline-flex rounded-full border border-cyan-900 bg-cyan-950 px-2 py-1 text-xs font-medium text-cyan-300"
+ end
+ end
+end
diff --git a/lib/reencodarr_web/router.ex b/lib/reencodarr_web/router.ex
index aa2840e6..4326f337 100644
--- a/lib/reencodarr_web/router.ex
+++ b/lib/reencodarr_web/router.ex
@@ -30,6 +30,7 @@ defmodule ReencodarrWeb.Router do
live_session :default, layout: {ReencodarrWeb.Layouts, :app} do
live "/", DashboardLive, :index
+ live "/workers", WorkersLive, :index
live "/broadway", BroadwayLive, :index
live "/bad-files", BadFilesLive, :index
live "/failures", FailuresLive, :index
@@ -60,6 +61,10 @@ defmodule ReencodarrWeb.Router do
post "/webhooks/radarr", RadarrWebhookController, :radarr
end
+ scope "/workers", ReencodarrWeb do
+ get "/files/:id", WorkerFileController, :show
+ end
+
# Other scopes may use custom stacks.
# scope "/api", ReencodarrWeb do
# pipe_through :api
diff --git a/nix/dev-shell.nix b/nix/dev-shell.nix
index bd1b58da..78146b4b 100644
--- a/nix/dev-shell.nix
+++ b/nix/dev-shell.nix
@@ -72,6 +72,11 @@ in
export MIX_OS_DEPS_COMPILE_PARTITION_COUNT=$(( $(nproc) / 2 ))
export ERL_AFLAGS="-kernel shell_history enabled"
export SECRET_KEY_BASE="WEWsPGIpK/OgJA2ZcwzsgZxWKSAp35IsqWPYsvSUmm5awBUGpvsVOcG2kkDteXR1"
+ export REENCODARR_WORKER_TOKEN="$(printf '%s' "''${SECRET_KEY_BASE}:ab-av1-worker-token" \
+ | ${pkgs.openssl}/bin/openssl dgst -sha256 -binary \
+ | ${pkgs.coreutils}/bin/base64 -w 0 \
+ | ${pkgs.coreutils}/bin/tr '+/' '-_' \
+ | ${pkgs.coreutils}/bin/tr -d '=')"
export COMPOSE_BAKE=true
export GPG_TTY=$(tty)
diff --git a/nix/module.nix b/nix/module.nix
index caeb048e..2a20c339 100644
--- a/nix/module.nix
+++ b/nix/module.nix
@@ -30,6 +30,15 @@
${lib.optionalString (cfg.secretKeyBaseFile != null) ''
export SECRET_KEY_BASE="$(< "$CREDENTIALS_DIRECTORY/secret_key_base")"
''}
+ if [ -z "''${REENCODARR_WORKER_TOKEN:-}" ] && [ -n "''${SECRET_KEY_BASE:-}" ]; then
+ export REENCODARR_WORKER_TOKEN="$(
+ printf '%s' "''${SECRET_KEY_BASE}:ab-av1-worker-token" \
+ | ${pkgs.openssl}/bin/openssl dgst -sha256 -binary \
+ | ${pkgs.coreutils}/bin/base64 -w 0 \
+ | ${pkgs.coreutils}/bin/tr '+/' '-_' \
+ | ${pkgs.coreutils}/bin/tr -d '='
+ )"
+ fi
'';
startScript = pkgs.writeShellScript "reencodarr-start" ''
diff --git a/priv/repo/migrations/20260708000000_add_crf_search_worker_id_to_videos.exs b/priv/repo/migrations/20260708000000_add_crf_search_worker_id_to_videos.exs
new file mode 100644
index 00000000..569eb9c0
--- /dev/null
+++ b/priv/repo/migrations/20260708000000_add_crf_search_worker_id_to_videos.exs
@@ -0,0 +1,11 @@
+defmodule Reencodarr.Repo.Migrations.AddCrfSearchWorkerIdToVideos do
+ use Ecto.Migration
+
+ def change do
+ alter table(:videos) do
+ add :crf_search_worker_id, :string
+ end
+
+ create index(:videos, [:state, :crf_search_worker_id])
+ end
+end
diff --git a/test/reencodarr/ab_av1/crf_search_vmaf_retry_test.exs b/test/reencodarr/ab_av1/crf_search_vmaf_retry_test.exs
index 8701edee..ab24855b 100644
--- a/test/reencodarr/ab_av1/crf_search_vmaf_retry_test.exs
+++ b/test/reencodarr/ab_av1/crf_search_vmaf_retry_test.exs
@@ -97,7 +97,9 @@ defmodule Reencodarr.AbAv1.CrfSearchVmafRetryTest do
send(test_pid, {:open_port_call, count, args})
# Spawn a script that outputs lines then exits with failure
- script = Enum.map_join(output_lines, "; ", &"echo '#{&1}'") <> "; exit 1"
+ # Keep the port alive briefly so CrfSearch can finish its startup call
+ # chain before the mocked failure process exits.
+ script = Enum.map_join(output_lines, "; ", &"echo '#{&1}'") <> "; sleep 0.1; exit 1"
port = Port.open({:spawn, "sh -c \"#{script}\""}, [:exit_status, :binary, {:line, 1024}])
{:ok, port}
end)
diff --git a/test/reencodarr/ab_av1/worker_config_test.exs b/test/reencodarr/ab_av1/worker_config_test.exs
new file mode 100644
index 00000000..8b0ac16a
--- /dev/null
+++ b/test/reencodarr/ab_av1/worker_config_test.exs
@@ -0,0 +1,23 @@
+defmodule Reencodarr.AbAv1.WorkerConfigTest do
+ use ExUnit.Case, async: false
+
+ alias Reencodarr.AbAv1.WorkerConfig
+
+ test "can enable and disable distributed workers" do
+ previous = Application.get_env(:reencodarr, :distributed_worker_enabled)
+
+ try do
+ assert :ok = WorkerConfig.enable()
+ assert WorkerConfig.enabled?()
+
+ assert :ok = WorkerConfig.disable()
+ refute WorkerConfig.enabled?()
+ after
+ if is_nil(previous) do
+ Application.delete_env(:reencodarr, :distributed_worker_enabled)
+ else
+ Application.put_env(:reencodarr, :distributed_worker_enabled, previous)
+ end
+ end
+ end
+end
diff --git a/test/reencodarr/ab_av1/worker_job_state_machine_test.exs b/test/reencodarr/ab_av1/worker_job_state_machine_test.exs
new file mode 100644
index 00000000..3fdddd54
--- /dev/null
+++ b/test/reencodarr/ab_av1/worker_job_state_machine_test.exs
@@ -0,0 +1,123 @@
+defmodule Reencodarr.AbAv1.WorkerJobStateMachineTest do
+ use ExUnit.Case, async: true
+
+ alias Reencodarr.AbAv1.WorkerJobStateMachine
+
+ test "complete transfer progress moves the worker to input ready" do
+ session = session()
+
+ assert {:ok, session} = WorkerJobStateMachine.assign_video(session, 123, :receiving_input)
+ assert session.phase == :receiving_input
+ assert session.active_video_id == 123
+
+ assert {:ok, session} =
+ WorkerJobStateMachine.set_transfer_progress(session, %{
+ video_id: 123,
+ percent: 100.0
+ })
+
+ assert session.phase == :input_ready
+ assert session.transfer_progress.percent == 100.0
+
+ assert {:ok, session} = WorkerJobStateMachine.finish_transfer(session)
+ assert session.phase == :input_ready
+ assert session.active_video_id == 123
+ assert session.transfer_progress.percent == 100.0
+ end
+
+ test "CRF progress wins over stale transfer progress" do
+ session =
+ session()
+ |> Map.merge(%{
+ phase: :receiving_input,
+ active_video_id: 123,
+ transfer_progress: %{video_id: 123, percent: 100.0}
+ })
+
+ assert {:ok, session} =
+ WorkerJobStateMachine.set_crf_search_progress(session, %{
+ video_id: 123,
+ percent: 75.0
+ })
+
+ assert session.phase == :crf_searching
+ assert session.active_video_id == 123
+ assert is_nil(session.transfer_progress)
+ assert session.crf_search_progress.percent == 75.0
+ end
+
+ test "progress cannot move an active worker to a different video" do
+ session =
+ session()
+ |> Map.merge(%{
+ phase: :receiving_input,
+ active_video_id: 123,
+ transfer_progress: %{video_id: 123, percent: 50.0}
+ })
+
+ assert {:error, :invalid_worker_phase} =
+ WorkerJobStateMachine.set_transfer_progress(session, %{
+ video_id: 456,
+ percent: 75.0
+ })
+
+ assert {:error, :invalid_worker_phase} =
+ WorkerJobStateMachine.set_crf_search_progress(session, %{
+ video_id: 456,
+ percent: 10.0
+ })
+ end
+
+ test "transfer progress for the active video moves CRF search back to receiving input" do
+ session =
+ session()
+ |> Map.merge(%{
+ phase: :crf_searching,
+ active_video_id: 123,
+ crf_search_progress: %{video_id: 123, percent: 25.0}
+ })
+
+ assert {:ok, session} =
+ WorkerJobStateMachine.set_transfer_progress(session, %{
+ video_id: 123,
+ percent: 25.0
+ })
+
+ assert session.phase == :receiving_input
+ assert session.active_video_id == 123
+ assert session.transfer_progress.percent == 25.0
+ assert is_nil(session.crf_search_progress)
+ end
+
+ test "clearing active work is the only transition that makes the worker idle" do
+ session =
+ session()
+ |> Map.merge(%{
+ phase: :crf_searching,
+ active_video_id: 123,
+ crf_search_progress: %{video_id: 123, percent: 75.0}
+ })
+
+ assert {:ok, session} = WorkerJobStateMachine.clear_video(session)
+ assert session.phase == :idle
+ assert is_nil(session.active_video_id)
+ assert is_nil(session.crf_search_progress)
+ end
+
+ defp session do
+ %{
+ server_worker_id: "worker-server-1",
+ client_worker_id: "worker-client-1",
+ version: "0.11.4",
+ protocol_version: 1,
+ capabilities: %{"crf_search" => true},
+ phase: :idle,
+ active_video_id: nil,
+ transfer_progress: nil,
+ crf_search_progress: nil,
+ resource_usage: nil,
+ connected_at: DateTime.utc_now(),
+ last_seen_at: DateTime.utc_now()
+ }
+ end
+end
diff --git a/test/reencodarr/ab_av1/worker_protocol_test.exs b/test/reencodarr/ab_av1/worker_protocol_test.exs
new file mode 100644
index 00000000..75d82418
--- /dev/null
+++ b/test/reencodarr/ab_av1/worker_protocol_test.exs
@@ -0,0 +1,331 @@
+defmodule Reencodarr.AbAv1.WorkerProtocolTest do
+ use ExUnit.Case, async: false
+
+ alias Reencodarr.AbAv1.WorkerProtocol
+ alias Reencodarr.AbAv1.WorkerProtocol.Announcement
+ alias Reencodarr.AbAv1.WorkerProtocol.Completion
+ alias Reencodarr.AbAv1.WorkerProtocol.CrfSearchProgress
+ alias Reencodarr.AbAv1.WorkerProtocol.CrfSearchResult
+ alias Reencodarr.AbAv1.WorkerProtocol.FailureReport
+ alias Reencodarr.AbAv1.WorkerProtocol.TransferProgress
+
+ test "rejects invalid announcement payloads" do
+ assert {:error, :invalid_announcement} = WorkerProtocol.parse_announcement(%{})
+
+ assert {:error, :invalid_announcement} =
+ WorkerProtocol.parse_announcement(%{"worker_id" => "x"})
+ end
+
+ test "parses a worker announcement into a typed struct" do
+ assert {:ok,
+ %Announcement{
+ worker_id: "abav1-dev",
+ protocol_version: 1,
+ version: "0.10.0",
+ capabilities: %{"crf_search" => true}
+ }} =
+ WorkerProtocol.parse_announcement(%{
+ "worker_id" => "abav1-dev",
+ "protocol_version" => 1,
+ "version" => "0.10.0",
+ "capabilities" => %{"crf_search" => true}
+ })
+ end
+
+ test "maps protocol errors to wire payloads" do
+ assert WorkerProtocol.error(:unauthorized) == %{reason: "unauthorized"}
+ assert WorkerProtocol.error(:unsupported_event) == %{reason: "unsupported_event"}
+ end
+
+ test "parses transfer progress into a typed payload" do
+ assert {:ok,
+ %TransferProgress{
+ job_id: "job-1",
+ video_id: 123,
+ transfer_id: "job-1",
+ filename: "movie.mkv",
+ percent: 42.5,
+ bytes_sent: 1_048_576,
+ total_bytes: 2_097_152,
+ bytes_per_second: 524_288,
+ eta: 12,
+ chunk_index: 3,
+ total_chunks: 8
+ }} =
+ WorkerProtocol.parse_transfer_progress(%{
+ "video_id" => 123,
+ "job_id" => "job-1",
+ "percent" => 42.5,
+ "filename" => "movie.mkv",
+ "received_bytes" => 1_048_576,
+ "expected_bytes" => 2_097_152,
+ "bytes_per_second" => 524_288,
+ "eta" => 12,
+ "chunk_index" => 3,
+ "total_chunks" => 8
+ })
+ end
+
+ test "parses CRF search progress into a typed payload" do
+ assert {:ok,
+ %CrfSearchProgress{
+ video_id: 123,
+ percent: 78.5,
+ filename: "movie.mkv",
+ eta: 42,
+ fps: 23.97,
+ crf: 24.0,
+ sample_num: 2,
+ total_samples: 8
+ }} =
+ WorkerProtocol.parse_crf_search_progress(%{
+ "video_id" => 123,
+ "percent" => 78.5,
+ "filename" => "movie.mkv",
+ "eta" => 42,
+ "fps" => 23.97,
+ "crf" => 24.0,
+ "sample_num" => 2,
+ "total_samples" => 8
+ })
+ end
+
+ test "parses heartbeat resource usage" do
+ assert %{
+ cpu_percent: 87.5,
+ memory_bytes: 1_073_741_824,
+ memory_total_bytes: 4_294_967_296,
+ disk_free_bytes: 536_870_912_000,
+ disk_total_bytes: 1_099_511_627_776
+ } =
+ WorkerProtocol.parse_resource_usage(%{
+ "cpu_percent" => 87.5,
+ "memory_rss_bytes" => 1_073_741_824,
+ "total_memory_bytes" => 4_294_967_296,
+ "free_disk_bytes" => 536_870_912_000,
+ "total_disk_bytes" => 1_099_511_627_776
+ })
+
+ assert is_nil(WorkerProtocol.parse_resource_usage(%{}))
+ end
+
+ test "parses structured CRF results and completion payloads" do
+ assert {:ok,
+ %CrfSearchResult{
+ video_id: 123,
+ results: [
+ %{
+ crf: 28,
+ score: 95.4,
+ percent: 94,
+ size: "512 MB",
+ time: 3600,
+ params: ["--preset", "4"],
+ chosen: true
+ }
+ ]
+ }} =
+ WorkerProtocol.parse_crf_search_result(%{
+ "video_id" => 123,
+ "crf" => 28,
+ "score" => 95.4,
+ "percent" => 94,
+ "size" => "512 MB",
+ "time" => 3600,
+ "params" => ["--preset", "4"],
+ "chosen" => true
+ })
+
+ assert {:ok, %Completion{video_id: 123, result: :ok, chosen_crf: 28}} =
+ WorkerProtocol.parse_completion(%{
+ "video_id" => 123,
+ "result" => "ok",
+ "chosen_crf" => 28
+ })
+
+ assert {:ok,
+ %CrfSearchResult{
+ results: [%{chosen: false}]
+ }} =
+ WorkerProtocol.parse_crf_search_result(%{
+ "video_id" => 123,
+ "crf" => 28,
+ "score" => 95.4,
+ "percent" => 94,
+ "chosen" => false
+ })
+
+ assert {:ok,
+ %CrfSearchResult{
+ video_id: 123,
+ results: [
+ %{
+ crf: 31.5,
+ score: 96.2,
+ percent: 42.5,
+ size: "123456",
+ time: 88
+ }
+ ]
+ }} =
+ WorkerProtocol.parse_crf_search_result(%{
+ "job_id" => "job-123",
+ "video_id" => 123,
+ "source_name" => "movie.mkv",
+ "crf" => 31.5,
+ "vmaf_score" => 96.2,
+ "xpsnr_score" => nil,
+ "predicted_encode_size" => 123_456,
+ "encode_percent" => 42.5,
+ "predicted_encode_time_secs" => 87.5,
+ "from_cache" => false
+ })
+
+ assert {:ok,
+ %FailureReport{
+ retriable: false
+ }} =
+ WorkerProtocol.parse_failure_report(%{
+ "video_id" => 123,
+ "stage" => "crf_search",
+ "category" => "timeout",
+ "message" => "timed out",
+ "retriable" => false
+ })
+ end
+
+ test "parses typed failure reports" do
+ assert {:ok,
+ %FailureReport{
+ video_id: 123,
+ stage: :crf_search,
+ category: :timeout,
+ message: "timed out",
+ code: "EXIT_137",
+ context: %{node: "worker@host"},
+ retriable: true,
+ stderr_excerpt: "ab-av1 timed out"
+ }} =
+ WorkerProtocol.parse_failure_report(%{
+ "video_id" => 123,
+ "stage" => "crf_search",
+ "category" => "timeout",
+ "message" => "timed out",
+ "code" => "EXIT_137",
+ "context" => %{node: "worker@host"},
+ "retriable" => true,
+ "stderr_excerpt" => "ab-av1 timed out"
+ })
+ end
+
+ test "builds a job_assigned payload from the claimed video" do
+ payload =
+ WorkerProtocol.work_assigned(
+ %Reencodarr.Media.Video{
+ id: 123,
+ path: "/videos/movie.mkv",
+ size: 987_654
+ },
+ 96.5
+ )
+
+ assert %{
+ status: "job_assigned",
+ job_id: "123",
+ video_id: 123,
+ source_name: "movie.mkv",
+ size_bytes: 987_654,
+ chunk_size_bytes: 134_217_728,
+ target_vmaf: 96.5,
+ crf_search_args: crf_search_args
+ } = payload
+
+ assert [
+ "crf-search",
+ "--input",
+ "/videos/movie.mkv",
+ "--min-vmaf",
+ "96.5" | _rest
+ ] = crf_search_args
+
+ assert "--temp-dir" in crf_search_args
+ end
+
+ test "includes configured worker transfer URL in job payloads" do
+ previous_base_url = Application.get_env(:reencodarr, :worker_transfer_base_url)
+ previous_token = Application.get_env(:reencodarr, :worker_transfer_token)
+ Application.put_env(:reencodarr, :worker_transfer_base_url, "http://10.0.0.10:4000/")
+ Application.put_env(:reencodarr, :worker_transfer_token, "transfer-token")
+
+ on_exit(fn ->
+ restore_env(:worker_transfer_base_url, previous_base_url)
+ restore_env(:worker_transfer_token, previous_token)
+ end)
+
+ video = %Reencodarr.Media.Video{id: 123, path: "/videos/movie.mkv", size: 987_654}
+
+ assert %{transfer: transfer} = WorkerProtocol.work_assigned(video, 96.5)
+
+ assert transfer == %{
+ url: "http://10.0.0.10:4000/workers/files/123",
+ auth: %{
+ scheme: "bearer",
+ header: "authorization",
+ value: "Bearer transfer-token"
+ }
+ }
+
+ assert %{transfer: ^transfer} =
+ WorkerProtocol.work_assigned(video, 96.5)
+
+ assert %{transfer: ^transfer} =
+ WorkerProtocol.work_in_progress(video, 96.5)
+ end
+
+ defp restore_env(key, nil), do: Application.delete_env(:reencodarr, key)
+ defp restore_env(key, value), do: Application.put_env(:reencodarr, key, value)
+
+ test "builds binary transfer chunk frames with ordered metadata and raw data" do
+ chunk = "raw video bytes"
+
+ assert {:binary, frame} =
+ WorkerProtocol.transfer_chunk(
+ %Reencodarr.Media.Video{id: 123},
+ "transfer-123",
+ 7,
+ 9,
+ 1_024,
+ 2_048,
+ chunk
+ )
+
+ assert {:ok,
+ %{
+ video_id: 123,
+ transfer_id: "transfer-123",
+ chunk_index: 7,
+ total_chunks: 9,
+ bytes_sent: 1_024,
+ total_bytes: 2_048,
+ crc32: crc32,
+ data: ^chunk
+ }} = WorkerProtocol.parse_transfer_chunk_frame(frame)
+
+ assert crc32 == :erlang.crc32(chunk)
+ end
+
+ test "reads chunk size from configuration" do
+ previous = Application.get_env(:reencodarr, :worker_chunk_size_bytes)
+ Application.put_env(:reencodarr, :worker_chunk_size_bytes, 2_097_152)
+
+ try do
+ assert WorkerProtocol.chunk_size_bytes() == 2_097_152
+ after
+ if is_nil(previous) do
+ Application.delete_env(:reencodarr, :worker_chunk_size_bytes)
+ else
+ Application.put_env(:reencodarr, :worker_chunk_size_bytes, previous)
+ end
+ end
+ end
+end
diff --git a/test/reencodarr/ab_av1/worker_sessions_test.exs b/test/reencodarr/ab_av1/worker_sessions_test.exs
new file mode 100644
index 00000000..71404242
--- /dev/null
+++ b/test/reencodarr/ab_av1/worker_sessions_test.exs
@@ -0,0 +1,401 @@
+defmodule Reencodarr.AbAv1.WorkerSessionsTest do
+ use Reencodarr.DataCase, async: false
+
+ alias Reencodarr.AbAv1.WorkerSessions
+ alias Reencodarr.Dashboard.Events
+ alias Reencodarr.Diagnostics
+ alias Reencodarr.Fixtures
+ alias Reencodarr.Media
+
+ setup do
+ WorkerSessions.reset()
+ :ok
+ end
+
+ test "rejects duplicate worker ids" do
+ assert {:ok, _session} = WorkerSessions.register(worker_session_attrs())
+
+ assert {:error, :duplicate_worker_id} =
+ WorkerSessions.register(worker_session_attrs(client_worker_id: "worker-client-2"))
+ end
+
+ test "returns unknown worker session for missing ids" do
+ assert {:error, :unknown_worker_session} = WorkerSessions.touch("missing-worker")
+ assert {:error, :unknown_worker_session} = WorkerSessions.assign_video("missing-worker", 123)
+ assert {:error, :unknown_worker_session} = WorkerSessions.clear_video("missing-worker")
+ assert {:error, :unknown_worker_session} = WorkerSessions.finish_transfer("missing-worker")
+ end
+
+ test "expires stale worker sessions" do
+ assert {:ok, _session} = WorkerSessions.register(worker_session_attrs())
+
+ assert [_session] = WorkerSessions.list()
+
+ assert {:ok, expired_sessions} = WorkerSessions.expire_stale(0)
+ assert Enum.map(expired_sessions, & &1.client_worker_id) == ["worker-client-1"]
+ assert WorkerSessions.list() == []
+ end
+
+ test "expires stale active sessions by requeueing only in-progress videos" do
+ {:ok, active_video} = Fixtures.video_fixture(%{state: :crf_searching})
+
+ {:ok, dispatched_video} =
+ Fixtures.video_fixture(%{state: :crf_searching, crf_search_worker_id: "worker-client-3"})
+
+ {:ok, completed_video} = Fixtures.video_fixture(%{state: :crf_searched})
+ _vmaf = Fixtures.vmaf_fixture(%{video_id: completed_video.id, crf: 28.0, score: 96.4})
+ assert {:ok, _} = Media.mark_vmaf_as_chosen(completed_video.id, 28.0)
+
+ assert {:ok, _session} =
+ WorkerSessions.register(worker_session_attrs(server_worker_id: "worker-server-1"))
+
+ assert {:ok, _session} =
+ WorkerSessions.register(
+ worker_session_attrs(
+ server_worker_id: "worker-server-2",
+ client_worker_id: "worker-client-2"
+ )
+ )
+
+ assert {:ok, _session} =
+ WorkerSessions.register(
+ worker_session_attrs(
+ server_worker_id: "worker-server-3",
+ client_worker_id: "worker-client-3"
+ )
+ )
+
+ assert {:ok, _session} = WorkerSessions.assign_video("worker-server-1", active_video.id)
+ assert {:ok, _session} = WorkerSessions.assign_video("worker-server-2", completed_video.id)
+ assert {:ok, _session} = WorkerSessions.assign_video("worker-server-3", dispatched_video.id)
+
+ assert {:ok, [_session_one, _session_two, _session_three]} = WorkerSessions.expire_stale(0)
+ assert Media.get_video(active_video.id).state == :analyzed
+ assert Media.get_video(dispatched_video.id).state == :crf_searching
+ assert Media.get_video(completed_video.id).state == :crf_searched
+ assert WorkerSessions.list() == []
+ end
+
+ test "timer-driven stale expiry removes old sessions" do
+ previous_timeout = Application.get_env(:reencodarr, :worker_session_timeout_seconds)
+ Application.put_env(:reencodarr, :worker_session_timeout_seconds, 0)
+
+ on_exit(fn ->
+ Application.put_env(:reencodarr, :worker_session_timeout_seconds, previous_timeout)
+ end)
+
+ Phoenix.PubSub.subscribe(Reencodarr.PubSub, Events.channel())
+ assert {:ok, _session} = WorkerSessions.register(worker_session_attrs())
+
+ send(Process.whereis(WorkerSessions), :expire_stale)
+ assert_receive {:worker_sessions_updated, %{sessions: []}}
+
+ assert WorkerSessions.list() == []
+ end
+
+ test "includes worker sessions in diagnostics output" do
+ assert {:ok, _session} = WorkerSessions.register(worker_session_attrs())
+ {:ok, video} = Fixtures.video_fixture(%{state: :analyzed})
+ assert {:ok, _session} = WorkerSessions.assign_video("worker-server-1", video.id)
+
+ output = Diagnostics.processes()
+
+ assert output =~ "Worker Sessions:"
+ assert output =~ "worker-client-1"
+ assert output =~ "protocol=1"
+ assert output =~ "version=0.10.0"
+ assert output =~ "phase=crf_searching"
+ assert output =~ "video_state=analyzed"
+ end
+
+ test "tracks an assigned video on the session" do
+ assert {:ok, _session} = WorkerSessions.register(worker_session_attrs())
+
+ assert {:ok, session} = WorkerSessions.assign_video("worker-server-1", 123)
+ assert session.active_video_id == 123
+ assert session.phase == :crf_searching
+
+ [listed_session] = WorkerSessions.list()
+ assert listed_session.active_video_id == 123
+ assert listed_session.phase == :crf_searching
+ end
+
+ test "complete transfer progress moves to input ready without clearing the active video" do
+ assert {:ok, _session} = WorkerSessions.register(worker_session_attrs())
+
+ assert {:ok, session} = WorkerSessions.assign_video("worker-server-1", 123, :receiving_input)
+ assert session.active_video_id == 123
+ assert session.phase == :receiving_input
+
+ assert {:ok, session} =
+ WorkerSessions.set_transfer_progress("worker-server-1", %{
+ job_id: "job-1",
+ video_id: 123,
+ transfer_id: "job-1",
+ filename: "sample.mkv",
+ percent: 100.0,
+ bytes_sent: 10_485_760,
+ total_bytes: 10_485_760
+ })
+
+ assert session.active_video_id == 123
+ assert session.phase == :input_ready
+ assert session.transfer_progress.percent == 100.0
+
+ assert {:ok, session} = WorkerSessions.finish_transfer("worker-server-1")
+ assert session.active_video_id == 123
+ assert session.phase == :input_ready
+ assert session.transfer_progress.percent == 100.0
+ end
+
+ test "rejects progress for a different active video" do
+ assert {:ok, _session} = WorkerSessions.register(worker_session_attrs())
+ assert {:ok, _session} = WorkerSessions.assign_video("worker-server-1", 123, :receiving_input)
+
+ assert {:error, :invalid_worker_phase} =
+ WorkerSessions.set_transfer_progress("worker-server-1", %{
+ video_id: 456,
+ percent: 25.0
+ })
+
+ assert {:error, :invalid_worker_phase} =
+ WorkerSessions.set_crf_search_progress("worker-server-1", %{
+ video_id: 456,
+ percent: 25.0
+ })
+
+ session = WorkerSessions.get("worker-server-1")
+ assert session.active_video_id == 123
+ assert session.phase == :receiving_input
+ assert is_nil(session.transfer_progress)
+ assert is_nil(session.crf_search_progress)
+ end
+
+ test "keeps CRF sample metadata when later progress omits it" do
+ assert {:ok, _session} = WorkerSessions.register(worker_session_attrs())
+
+ assert {:ok, session} =
+ WorkerSessions.set_crf_search_progress("worker-server-1", %{
+ video_id: 123,
+ percent: 10.0,
+ fps: 24.0,
+ crf: 28.0,
+ sample_num: 3,
+ total_samples: 5
+ })
+
+ assert session.crf_search_progress.crf == 28.0
+ assert session.crf_search_progress.sample_num == 3
+ assert session.crf_search_progress.total_samples == 5
+
+ assert {:ok, session} =
+ WorkerSessions.set_crf_search_progress("worker-server-1", %{
+ video_id: 123,
+ percent: 25.0,
+ fps: 25.0,
+ crf: nil,
+ sample_num: nil,
+ total_samples: nil
+ })
+
+ assert session.crf_search_progress.percent == 25.0
+ assert session.crf_search_progress.fps == 25.0
+ assert session.crf_search_progress.crf == 28.0
+ assert session.crf_search_progress.sample_num == 3
+ assert session.crf_search_progress.total_samples == 5
+ end
+
+ test "touches session when transfer progress is reported" do
+ assert {:ok, _session} = WorkerSessions.register(worker_session_attrs())
+
+ initial_session = WorkerSessions.get("worker-server-1")
+ initial_last_seen = initial_session.last_seen_at
+
+ Process.sleep(1_100)
+
+ assert {:ok, session} =
+ WorkerSessions.set_transfer_progress("worker-server-1", %{
+ job_id: "job-1",
+ video_id: 123,
+ transfer_id: "job-1",
+ filename: "sample.mkv",
+ percent: 25.0,
+ bytes_sent: 2_621_440,
+ total_bytes: 10_485_760,
+ bytes_per_second: 1_048_576,
+ eta: 15,
+ chunk_index: 2,
+ total_chunks: 8
+ })
+
+ assert DateTime.compare(session.last_seen_at, initial_last_seen) == :gt
+ end
+
+ test "touches session when CRF search progress is reported" do
+ assert {:ok, _session} = WorkerSessions.register(worker_session_attrs())
+
+ initial_session = WorkerSessions.get("worker-server-1")
+ initial_last_seen = initial_session.last_seen_at
+
+ Process.sleep(1_100)
+
+ assert {:ok, session} =
+ WorkerSessions.set_crf_search_progress("worker-server-1", %{
+ video_id: 123,
+ percent: 25.0,
+ fps: 24.0,
+ crf: 28.0,
+ sample_num: 3,
+ total_samples: 5
+ })
+
+ assert DateTime.compare(session.last_seen_at, initial_last_seen) == :gt
+ end
+
+ test "replaces reconnecting client sessions without dropping active state" do
+ {:ok, video} = Fixtures.video_fixture(%{state: :crf_searching})
+
+ assert {:ok, _session} =
+ WorkerSessions.register(
+ worker_session_attrs(
+ server_worker_id: "worker-server-1",
+ client_worker_id: "worker-client-1"
+ )
+ )
+
+ assert {:ok, _session} = WorkerSessions.assign_video("worker-server-1", video.id)
+
+ assert {:ok, session} =
+ WorkerSessions.set_crf_search_progress("worker-server-1", %{
+ video_id: video.id,
+ percent: 25.0,
+ fps: 24.0,
+ crf: 28.0,
+ sample_num: 2,
+ total_samples: 5
+ })
+
+ assert session.active_video_id == video.id
+ assert session.phase == :crf_searching
+
+ assert {:ok, session} =
+ WorkerSessions.register(
+ worker_session_attrs(
+ server_worker_id: "worker-server-2",
+ client_worker_id: "worker-client-1"
+ )
+ )
+
+ assert session.server_worker_id == "worker-server-2"
+ assert session.client_worker_id == "worker-client-1"
+ assert session.active_video_id == video.id
+ assert session.phase == :crf_searching
+ assert session.crf_search_progress.video_id == video.id
+ assert session.crf_search_progress.sample_num == 2
+ assert is_nil(WorkerSessions.get("worker-server-1"))
+ end
+
+ test "does not preserve completed active state when replacing reconnecting sessions" do
+ {:ok, video} = Fixtures.video_fixture(%{state: :crf_searched})
+ _vmaf = Fixtures.vmaf_fixture(%{video_id: video.id, crf: 28.0, score: 96.4})
+ assert {:ok, _} = Media.mark_vmaf_as_chosen(video.id, 28.0)
+
+ assert {:ok, _session} =
+ WorkerSessions.register(
+ worker_session_attrs(
+ server_worker_id: "worker-server-1",
+ client_worker_id: "worker-client-1"
+ )
+ )
+
+ assert {:ok, _session} = WorkerSessions.assign_video("worker-server-1", video.id)
+
+ assert {:ok, _session} =
+ WorkerSessions.set_crf_search_progress("worker-server-1", %{
+ video_id: video.id,
+ percent: 100.0,
+ fps: 24.0
+ })
+
+ assert {:ok, session} =
+ WorkerSessions.register(
+ worker_session_attrs(
+ server_worker_id: "worker-server-2",
+ client_worker_id: "worker-client-1"
+ )
+ )
+
+ assert is_nil(session.active_video_id)
+ assert session.phase == :idle
+ assert is_nil(session.crf_search_progress)
+ assert Media.get_video(video.id).state == :crf_searched
+ end
+
+ test "cancels and drains active distributed work" do
+ {:ok, video_one} = Fixtures.video_fixture(%{state: :analyzed})
+ {:ok, video_two} = Fixtures.video_fixture(%{state: :analyzed})
+
+ assert {:ok, _session} =
+ WorkerSessions.register(worker_session_attrs(server_worker_id: "worker-server-1"))
+
+ assert {:ok, _session} =
+ WorkerSessions.register(
+ worker_session_attrs(
+ server_worker_id: "worker-server-2",
+ client_worker_id: "worker-client-2"
+ )
+ )
+
+ assert {:ok, _session} = WorkerSessions.assign_video("worker-server-1", video_one.id)
+ assert {:ok, _session} = WorkerSessions.assign_video("worker-server-2", video_two.id)
+
+ assert {:ok, _session} = WorkerSessions.cancel("worker-server-1")
+ assert Media.get_video(video_one.id).state == :analyzed
+
+ assert {:ok, drained_sessions} = WorkerSessions.drain()
+ assert Enum.map(drained_sessions, & &1.server_worker_id) == ["worker-server-2"]
+ assert Media.get_video(video_two.id).state == :analyzed
+ assert WorkerSessions.list() == []
+ end
+
+ test "cancel and drain do not reset completed videos" do
+ {:ok, video_one} = Fixtures.video_fixture(%{state: :crf_searched})
+ {:ok, video_two} = Fixtures.video_fixture(%{state: :crf_searched})
+ _vmaf_one = Fixtures.vmaf_fixture(%{video_id: video_one.id, crf: 28.0, score: 96.4})
+ _vmaf_two = Fixtures.vmaf_fixture(%{video_id: video_two.id, crf: 28.0, score: 96.4})
+ assert {:ok, _} = Media.mark_vmaf_as_chosen(video_one.id, 28.0)
+ assert {:ok, _} = Media.mark_vmaf_as_chosen(video_two.id, 28.0)
+
+ assert {:ok, _session} =
+ WorkerSessions.register(worker_session_attrs(server_worker_id: "worker-server-1"))
+
+ assert {:ok, _session} =
+ WorkerSessions.register(
+ worker_session_attrs(
+ server_worker_id: "worker-server-2",
+ client_worker_id: "worker-client-2"
+ )
+ )
+
+ assert {:ok, _session} = WorkerSessions.assign_video("worker-server-1", video_one.id)
+ assert {:ok, _session} = WorkerSessions.assign_video("worker-server-2", video_two.id)
+
+ assert {:ok, _session} = WorkerSessions.cancel("worker-server-1")
+ assert Media.get_video(video_one.id).state == :crf_searched
+
+ assert {:ok, [_session]} = WorkerSessions.drain()
+ assert Media.get_video(video_two.id).state == :crf_searched
+ assert WorkerSessions.list() == []
+ end
+
+ defp worker_session_attrs(overrides \\ []) do
+ %{
+ server_worker_id: "worker-server-1",
+ client_worker_id: "worker-client-1",
+ protocol_version: 1,
+ version: "0.10.0",
+ capabilities: %{"crf_search" => true}
+ }
+ |> Map.merge(Map.new(overrides))
+ end
+end
diff --git a/test/reencodarr/broadway_producers_test.exs b/test/reencodarr/broadway_producers_test.exs
index 895b1ad6..06388415 100644
--- a/test/reencodarr/broadway_producers_test.exs
+++ b/test/reencodarr/broadway_producers_test.exs
@@ -116,6 +116,53 @@ defmodule Reencodarr.BroadwayProducersTest do
assert is_list(videos)
end
+ test "atomic claim transitions video to :crf_searching" do
+ expect_crf_available()
+
+ {:ok, video} = video_fixture(%{state: :analyzed})
+
+ state = init_state(CrfProducer)
+ {:noreply, videos, _state2} = CrfProducer.handle_demand(1, state)
+
+ assert Enum.count(videos) == 1
+ assert hd(videos).id == video.id
+
+ refreshed = Reencodarr.Repo.get!(Reencodarr.Media.Video, video.id)
+ assert refreshed.state == :crf_searching
+ end
+
+ test "atomic claim broadcasts video mutation for dashboard counts" do
+ Phoenix.PubSub.subscribe(Reencodarr.PubSub, "video_state_transitions")
+ expect_crf_available()
+ {:ok, video} = video_fixture(%{state: :analyzed})
+
+ state = init_state(CrfProducer)
+ {:noreply, [_claimed], _state2} = CrfProducer.handle_demand(1, state)
+
+ assert_receive {:video_mutated,
+ %{
+ action: :update,
+ old_video: %{id: old_id, state: :analyzed},
+ new_video: %{id: new_id, state: :crf_searching}
+ }}
+
+ assert old_id == video.id
+ assert new_id == video.id
+ end
+
+ test "claimed CRF videos are not dispatched again" do
+ expect_crf_available()
+
+ {:ok, _video} = video_fixture(%{state: :analyzed})
+
+ state = init_state(CrfProducer)
+ {:noreply, videos1, state2} = CrfProducer.handle_demand(1, state)
+ assert Enum.count(videos1) == 1
+
+ {:noreply, videos2, _state3} = CrfProducer.handle_demand(1, state2)
+ assert videos2 == []
+ end
+
test "poll returns list when called" do
{:ok, _video} = video_fixture(%{state: :analyzed})
@@ -168,4 +215,17 @@ defmodule Reencodarr.BroadwayProducersTest do
assert {:noreply, [], ^state} = EncoderProducer.handle_info(:unknown, state)
end
end
+
+ defp expect_crf_available do
+ :meck.new(Reencodarr.AbAv1.CrfSearch, [:passthrough])
+ :meck.expect(Reencodarr.AbAv1.CrfSearch, :available?, fn -> :available end)
+
+ on_exit(fn ->
+ try do
+ :meck.unload(Reencodarr.AbAv1.CrfSearch)
+ catch
+ :error, {:not_mocked, Reencodarr.AbAv1.CrfSearch} -> :ok
+ end
+ end)
+ end
end
diff --git a/test/reencodarr/media/orphan_reset_test.exs b/test/reencodarr/media/orphan_reset_test.exs
index 001d42c2..211b7f35 100644
--- a/test/reencodarr/media/orphan_reset_test.exs
+++ b/test/reencodarr/media/orphan_reset_test.exs
@@ -22,6 +22,23 @@ defmodule Reencodarr.Media.OrphanResetTest do
assert updated.state == :analyzed
end
+ test "does not reset crf_searching videos dispatched to a worker" do
+ {:ok, video} =
+ Fixtures.video_fixture(%{
+ path: "/test/worker_crf.mkv",
+ state: :crf_searching,
+ crf_search_worker_id: "worker-a",
+ video_codecs: ["h264"],
+ audio_codecs: ["aac"]
+ })
+
+ assert :ok = Media.reset_orphaned_crf_searching()
+
+ updated = Media.get_video(video.id)
+ assert updated.state == :crf_searching
+ assert updated.crf_search_worker_id == "worker-a"
+ end
+
test "does not affect videos in other states" do
{:ok, analyzed} =
Fixtures.video_fixture(%{
diff --git a/test/reencodarr_web/channels/worker_channel_test.exs b/test/reencodarr_web/channels/worker_channel_test.exs
new file mode 100644
index 00000000..c675a40c
--- /dev/null
+++ b/test/reencodarr_web/channels/worker_channel_test.exs
@@ -0,0 +1,1373 @@
+defmodule ReencodarrWeb.WorkerChannelTest do
+ use ReencodarrWeb.ChannelCase, async: false
+
+ alias Reencodarr.AbAv1.WorkerProtocol
+ alias Reencodarr.AbAv1.WorkerSessions
+ alias Reencodarr.Dashboard.Events
+ alias Reencodarr.Fixtures
+ alias Reencodarr.Media
+ alias ReencodarrWeb.WorkerChannel
+ alias ReencodarrWeb.WorkerSocket
+ import Reencodarr.TestHelpers
+
+ describe "ab-av1 worker websocket" do
+ setup do
+ previous_enabled = Application.get_env(:reencodarr, :distributed_worker_enabled)
+ Application.put_env(:reencodarr, :distributed_worker_enabled, true)
+ WorkerSessions.reset()
+
+ on_exit(fn ->
+ if is_nil(previous_enabled) do
+ Application.delete_env(:reencodarr, :distributed_worker_enabled)
+ else
+ Application.put_env(:reencodarr, :distributed_worker_enabled, previous_enabled)
+ end
+ end)
+
+ :ok
+ end
+
+ test "authenticates, announces capabilities, and returns no work" do
+ token = "test-worker-token"
+ Application.put_env(:reencodarr, :worker_token, token)
+
+ assert {:ok, socket} = connect(WorkerSocket, %{"token" => token})
+ assert socket.assigns.worker_id =~ "worker-"
+
+ assert {:ok, %{worker_id: worker_id}, socket} =
+ subscribe_and_join(socket, "workers:crf_search")
+
+ assert worker_id == socket.assigns.worker_id
+
+ assert_reply push(socket, "announce", %{
+ "worker_id" => "abav1-dev",
+ "protocol_version" => 1,
+ "version" => "0.10.0",
+ "capabilities" => %{"crf_search" => true}
+ }),
+ :ok,
+ %{accepted: true, protocol_version: 1}
+
+ assert_reply push(socket, "pull_work", %{}), :ok, %{status: "no_work"}
+ after
+ Application.delete_env(:reencodarr, :worker_token)
+ end
+
+ test "assigns one queued video to a worker and marks it crf_searching" do
+ token = "test-worker-token"
+ Application.put_env(:reencodarr, :worker_token, token)
+
+ {:ok, video} = Fixtures.video_fixture(%{state: :analyzed})
+
+ assert {:ok, socket} = connect(WorkerSocket, %{"token" => token})
+ assert {:ok, _join_payload, socket} = subscribe_and_join(socket, "workers:crf_search")
+
+ assert_reply push(socket, "announce", announce_payload(worker_id: "worker-a")),
+ :ok,
+ %{accepted: true, protocol_version: 1}
+
+ assert_reply push(socket, "pull_work", %{}),
+ :ok,
+ %{
+ status: "job_assigned",
+ job_id: job_id,
+ video_id: assigned_video_id,
+ source_name: source_name,
+ size_bytes: size_bytes,
+ chunk_size_bytes: 134_217_728,
+ crf_search_args: crf_search_args
+ }
+
+ assert job_id == Integer.to_string(video.id)
+ assert assigned_video_id == video.id
+ assert source_name == Path.basename(video.path)
+ assert size_bytes == video.size
+ assert "crf-search" in crf_search_args
+ assert "--input" in crf_search_args
+ assert video.path in crf_search_args
+ assert "--min-vmaf" in crf_search_args
+ assert "95" in crf_search_args
+ assert "--encoder" in crf_search_args
+ assert "svt-av1" in crf_search_args
+ assert Media.get_video(video.id).state == :crf_searching
+ assert Media.get_video(video.id).crf_search_worker_id == "worker-a"
+ assert [session] = WorkerSessions.list()
+ assert session.active_video_id == video.id
+ assert session.phase == :receiving_input
+ assert is_nil(session.transfer_progress)
+ after
+ Application.delete_env(:reencodarr, :worker_token)
+ end
+
+ test "rejects unsupported protocol versions" do
+ token = "test-worker-token"
+ Application.put_env(:reencodarr, :worker_token, token)
+
+ assert {:ok, socket} = connect(WorkerSocket, %{"token" => token})
+ assert {:ok, _join_payload, socket} = subscribe_and_join(socket, "workers:crf_search")
+
+ assert_reply push(socket, "announce", announce_payload(protocol_version: 99)),
+ :error,
+ %{reason: "unsupported_protocol_version", supported_protocol_versions: [1]}
+ after
+ Application.delete_env(:reencodarr, :worker_token)
+ end
+
+ test "replaces reconnecting worker sessions with the same client id" do
+ token = "test-worker-token"
+ Application.put_env(:reencodarr, :worker_token, token)
+
+ assert {:ok, socket1} = connect(WorkerSocket, %{"token" => token})
+ assert {:ok, _join_payload, socket1} = subscribe_and_join(socket1, "workers:crf_search")
+ server_worker_id1 = socket1.assigns.worker_id
+
+ assert_reply push(socket1, "announce", announce_payload()),
+ :ok,
+ %{accepted: true, protocol_version: 1}
+
+ assert {:ok, socket2} = connect(WorkerSocket, %{"token" => token})
+ assert {:ok, _join_payload, socket2} = subscribe_and_join(socket2, "workers:crf_search")
+ server_worker_id2 = socket2.assigns.worker_id
+
+ assert_reply push(socket2, "announce", announce_payload()),
+ :ok,
+ %{accepted: true, protocol_version: 1}
+
+ assert is_nil(WorkerSessions.get(server_worker_id1))
+ assert WorkerSessions.get(server_worker_id2).client_worker_id == "abav1-dev"
+ after
+ Application.delete_env(:reencodarr, :worker_token)
+ end
+
+ test "does not hand the same queued video to two workers" do
+ token = "test-worker-token"
+ Application.put_env(:reencodarr, :worker_token, token)
+
+ {:ok, video} = Fixtures.video_fixture(%{state: :analyzed})
+
+ assert {:ok, socket1} = connect(WorkerSocket, %{"token" => token})
+ assert {:ok, _join_payload, socket1} = subscribe_and_join(socket1, "workers:crf_search")
+
+ assert_reply push(socket1, "announce", announce_payload(worker_id: "worker-a")),
+ :ok,
+ %{accepted: true, protocol_version: 1}
+
+ assert {:ok, socket2} = connect(WorkerSocket, %{"token" => token})
+ assert {:ok, _join_payload, socket2} = subscribe_and_join(socket2, "workers:crf_search")
+
+ assert_reply push(socket2, "announce", announce_payload(worker_id: "worker-b")),
+ :ok,
+ %{accepted: true, protocol_version: 1}
+
+ assert_reply push(socket1, "pull_work", %{}),
+ :ok,
+ %{status: "job_assigned", video_id: assigned_video_id}
+
+ assert assigned_video_id == video.id
+
+ assert_reply push(socket2, "pull_work", %{}),
+ :ok,
+ %{status: "no_work"}
+ after
+ Application.delete_env(:reencodarr, :worker_token)
+ end
+
+ test "tracks heartbeat last-seen for announced sessions" do
+ token = "test-worker-token"
+ Application.put_env(:reencodarr, :worker_token, token)
+
+ assert {:ok, socket} = connect(WorkerSocket, %{"token" => token})
+ assert {:ok, _join_payload, socket} = subscribe_and_join(socket, "workers:crf_search")
+
+ assert_reply push(socket, "announce", announce_payload()), :ok, %{
+ accepted: true,
+ protocol_version: 1
+ }
+
+ [session] = WorkerSessions.list()
+ connected_at = session.connected_at
+ first_seen_at = session.last_seen_at
+
+ assert_reply push(socket, "heartbeat", %{
+ "cpu_percent" => 87.5,
+ "memory_bytes" => 1_073_741_824,
+ "memory_total_bytes" => 4_294_967_296,
+ "disk_free_bytes" => 536_870_912_000,
+ "disk_total_bytes" => 1_099_511_627_776
+ }),
+ :ok,
+ %{
+ accepted: true,
+ last_seen_at: last_seen_at
+ }
+
+ assert {:ok, parsed_last_seen_at, 0} = DateTime.from_iso8601(last_seen_at)
+
+ [updated_session] = WorkerSessions.list()
+ assert updated_session.connected_at == connected_at
+ assert DateTime.compare(updated_session.last_seen_at, first_seen_at) in [:eq, :gt]
+ assert updated_session.last_seen_at == parsed_last_seen_at
+ assert updated_session.resource_usage.cpu_percent == 87.5
+ assert updated_session.resource_usage.memory_bytes == 1_073_741_824
+ assert updated_session.resource_usage.memory_total_bytes == 4_294_967_296
+ assert updated_session.resource_usage.disk_free_bytes == 536_870_912_000
+ assert updated_session.resource_usage.disk_total_bytes == 1_099_511_627_776
+ after
+ Application.delete_env(:reencodarr, :worker_token)
+ end
+
+ test "tracks last-seen for pull_work requests on active sessions" do
+ token = "test-worker-token"
+ Application.put_env(:reencodarr, :worker_token, token)
+
+ {:ok, video} = Fixtures.video_fixture(%{state: :analyzed})
+ video_id = video.id
+
+ assert {:ok, socket} = connect(WorkerSocket, %{"token" => token})
+ assert {:ok, _join_payload, socket} = subscribe_and_join(socket, "workers:crf_search")
+
+ assert_reply push(socket, "announce", announce_payload(worker_id: "worker-a")), :ok, %{
+ accepted: true
+ }
+
+ assert_reply push(socket, "pull_work", %{}), :ok, %{
+ status: "job_assigned",
+ video_id: ^video_id
+ }
+
+ [session] = WorkerSessions.list()
+ first_seen_at = session.last_seen_at
+
+ Process.sleep(1_100)
+
+ assert_reply push(socket, "pull_work", %{}), :ok, %{status: status, video_id: ^video_id}
+ assert status in ["job_assigned", "job_in_progress"]
+
+ [updated_session] = WorkerSessions.list()
+ assert DateTime.compare(updated_session.last_seen_at, first_seen_at) == :gt
+ after
+ Application.delete_env(:reencodarr, :worker_token)
+ end
+
+ test "broadcasts transfer and CRF progress updates" do
+ token = "test-worker-token"
+ Application.put_env(:reencodarr, :worker_token, token)
+ Phoenix.PubSub.subscribe(Reencodarr.PubSub, Events.channel())
+
+ {:ok, video} = Fixtures.video_fixture(%{state: :analyzed})
+ video_id = video.id
+
+ assert {:ok, socket} = connect(WorkerSocket, %{"token" => token})
+ assert {:ok, _join_payload, socket} = subscribe_and_join(socket, "workers:crf_search")
+ server_worker_id = socket.assigns.worker_id
+
+ assert_reply push(socket, "announce", announce_payload(worker_id: "worker-a")),
+ :ok,
+ %{accepted: true, protocol_version: 1}
+
+ assert_reply push(socket, "pull_work", %{}),
+ :ok,
+ %{status: "job_assigned", video_id: ^video_id}
+
+ assert_reply push(socket, "transfer_progress", %{
+ "video_id" => video_id,
+ "job_id" => "job-1",
+ "filename" => Path.basename(video.path),
+ "percent" => 25.5,
+ "received_bytes" => 2_621_440,
+ "expected_bytes" => 10_485_760,
+ "bytes_per_second" => 1_048_576,
+ "eta" => 15,
+ "chunk_index" => 2,
+ "total_chunks" => 8
+ }),
+ :ok,
+ %{accepted: true, event: "transfer_progress"}
+
+ assert_receive {:transfer_progress,
+ %{
+ video_id: ^video_id,
+ transfer_id: "job-1",
+ job_id: "job-1",
+ filename: filename,
+ percent: 25.5,
+ bytes_sent: 2_621_440,
+ total_bytes: 10_485_760,
+ bytes_per_second: 1_048_576,
+ eta: 15,
+ chunk_index: 2,
+ total_chunks: 8
+ }}
+
+ assert filename == Path.basename(video.path)
+
+ Process.sleep(10)
+
+ assert_reply push(socket, "transfer_progress", %{
+ "video_id" => video_id,
+ "job_id" => "job-1",
+ "filename" => Path.basename(video.path),
+ "percent" => 50.0,
+ "received_bytes" => 5_242_880,
+ "expected_bytes" => 10_485_760,
+ "chunk_index" => 3,
+ "total_chunks" => 8
+ }),
+ :ok,
+ %{accepted: true, event: "transfer_progress"}
+
+ assert_receive {:transfer_progress,
+ %{
+ video_id: ^video_id,
+ bytes_sent: 5_242_880,
+ total_bytes: 10_485_760,
+ bytes_per_second: derived_rate,
+ eta: derived_eta
+ }}
+
+ assert is_integer(derived_rate)
+ assert derived_rate > 0
+ assert is_integer(derived_eta)
+ assert derived_eta >= 0
+
+ assert_reply push(socket, "crf_search_progress", %{
+ "video_id" => video_id,
+ "percent" => 62.0,
+ "filename" => Path.basename(video.path),
+ "eta" => 90,
+ "fps" => 12.5,
+ "crf" => 28.0,
+ "sample_num" => 3,
+ "total_samples" => 8
+ }),
+ :ok,
+ %{accepted: true, event: "crf_search_progress"}
+
+ assert_receive {:crf_search_progress,
+ %{
+ video_id: ^video_id,
+ percent: 62.0,
+ filename: filename,
+ eta: 90,
+ fps: 12.5,
+ crf: 28.0,
+ sample_num: 3,
+ total_samples: 8
+ }}
+
+ assert filename == Path.basename(video.path)
+ session = WorkerSessions.get(server_worker_id)
+ assert session.phase == :crf_searching
+ assert is_nil(session.transfer_progress)
+ assert session.crf_search_progress.percent == 62.0
+
+ Phoenix.PubSub.broadcast(
+ Reencodarr.PubSub,
+ WorkerChannel.worker_control_topic(server_worker_id),
+ {:worker_control, :pause}
+ )
+
+ assert_push "control", %{action: "pause", video_id: ^video_id}
+ after
+ Application.delete_env(:reencodarr, :worker_token)
+ end
+
+ test "accepts resumed CRF progress for already-searching video after reconnect" do
+ token = "test-worker-token"
+ Application.put_env(:reencodarr, :worker_token, token)
+
+ {:ok, video} =
+ Fixtures.video_fixture(%{state: :crf_searching, crf_search_worker_id: "worker-a"})
+
+ video_id = video.id
+
+ assert {:ok, socket} = connect(WorkerSocket, %{"token" => token})
+ assert {:ok, _join_payload, socket} = subscribe_and_join(socket, "workers:crf_search")
+ server_worker_id = socket.assigns.worker_id
+
+ assert_reply push(socket, "announce", announce_payload(worker_id: "worker-a")),
+ :ok,
+ %{accepted: true, protocol_version: 1}
+
+ session = WorkerSessions.get(server_worker_id)
+ assert is_nil(session.active_video_id)
+
+ assert_reply push(socket, "crf_search_progress", %{
+ "video_id" => video_id,
+ "percent" => 16.4,
+ "filename" => Path.basename(video.path),
+ "fps" => 24.0,
+ "crf" => 28.0,
+ "sample_num" => 2,
+ "total_samples" => 8
+ }),
+ :ok,
+ %{accepted: true, event: "crf_search_progress"}
+
+ session = WorkerSessions.get(server_worker_id)
+ assert session.active_video_id == video_id
+ assert session.crf_search_progress.sample_num == 2
+ assert Media.get_video(video_id).state == :crf_searching
+ after
+ Application.delete_env(:reencodarr, :worker_token)
+ end
+
+ test "does not restart transfer when reconnected worker asks for work already in progress" do
+ token = "test-worker-token"
+ Application.put_env(:reencodarr, :worker_token, token)
+
+ {:ok, video} =
+ Fixtures.video_fixture(%{state: :crf_searching, crf_search_worker_id: "worker-a"})
+
+ video_id = video.id
+
+ assert {:ok, socket} = connect(WorkerSocket, %{"token" => token})
+ assert {:ok, _join_payload, socket} = subscribe_and_join(socket, "workers:crf_search")
+ server_worker_id = socket.assigns.worker_id
+
+ assert_reply push(socket, "announce", announce_payload(worker_id: "worker-a")),
+ :ok,
+ %{accepted: true, protocol_version: 1}
+
+ assert_reply push(socket, "crf_search_progress", %{
+ "video_id" => video_id,
+ "percent" => 12.5,
+ "filename" => Path.basename(video.path),
+ "fps" => 24.0,
+ "crf" => 28.0,
+ "sample_num" => 1,
+ "total_samples" => 8
+ }),
+ :ok,
+ %{accepted: true, event: "crf_search_progress"}
+
+ assert {:ok, session} =
+ WorkerSessions.set_transfer_progress(server_worker_id, %{
+ job_id: Integer.to_string(video_id),
+ video_id: video_id,
+ transfer_id: Integer.to_string(video_id),
+ filename: Path.basename(video.path),
+ percent: 100.0,
+ bytes_sent: 8,
+ total_bytes: 8
+ })
+
+ assert session.phase == :input_ready
+ assert session.active_video_id == video_id
+ assert is_nil(session.crf_search_progress)
+
+ assert_reply push(socket, "pull_work", %{}),
+ :ok,
+ %{
+ status: "job_in_progress",
+ video_id: ^video_id,
+ target_vmaf: 95,
+ crf_search_args: crf_search_args
+ }
+
+ assert "crf-search" in crf_search_args
+ assert "--input" in crf_search_args
+ assert video.path in crf_search_args
+ assert "--min-vmaf" in crf_search_args
+ assert "95" in crf_search_args
+
+ refute_push "transfer_started", _, 50
+ refute_push "transfer_chunk", _, 50
+
+ session = WorkerSessions.get(server_worker_id)
+ assert session.active_video_id == video_id
+ assert Media.get_video(video_id).state == :crf_searching
+ after
+ Application.delete_env(:reencodarr, :worker_token)
+ end
+
+ test "reattaches dispatched work when reconnecting worker asks for work" do
+ token = "test-worker-token"
+ Application.put_env(:reencodarr, :worker_token, token)
+
+ with_temp_file("abcdefgh", ".mkv", fn path ->
+ {:ok, video} =
+ Fixtures.video_fixture(%{
+ path: path,
+ size: 8,
+ state: :crf_searching,
+ crf_search_worker_id: "worker-a"
+ })
+
+ video_id = video.id
+
+ assert {:ok, socket} = connect(WorkerSocket, %{"token" => token})
+ assert {:ok, _join_payload, socket} = subscribe_and_join(socket, "workers:crf_search")
+
+ assert_reply push(socket, "announce", announce_payload(worker_id: "worker-a")),
+ :ok,
+ %{accepted: true, protocol_version: 1}
+
+ assert_reply push(socket, "pull_work", %{}),
+ :ok,
+ %{status: "job_in_progress", video_id: ^video_id}
+
+ assert Media.get_video(video_id).state == :crf_searching
+ assert Media.get_video_failures(video_id) == []
+ assert WorkerSessions.get(socket.assigns.worker_id).phase == :crf_searching
+ refute_push "transfer_started", _, 50
+ end)
+ after
+ Application.delete_env(:reencodarr, :worker_token)
+ end
+
+ test "resends dispatched work when reconnecting worker reports missing input" do
+ token = "test-worker-token"
+ Application.put_env(:reencodarr, :worker_token, token)
+
+ with_temp_file("abcdefgh", ".mkv", fn path ->
+ {:ok, video} =
+ Fixtures.video_fixture(%{
+ path: path,
+ size: 8,
+ state: :crf_searching,
+ crf_search_worker_id: "worker-a"
+ })
+
+ video_id = video.id
+
+ assert {:ok, socket} = connect(WorkerSocket, %{"token" => token})
+ assert {:ok, _join_payload, socket} = subscribe_and_join(socket, "workers:crf_search")
+
+ assert_reply push(socket, "announce", announce_payload(worker_id: "worker-a")),
+ :ok,
+ %{accepted: true, protocol_version: 1}
+
+ assert_reply push(socket, "pull_work", %{"input_missing" => true}),
+ :ok,
+ %{
+ status: "job_assigned",
+ video_id: ^video_id,
+ crf_search_args: crf_search_args
+ }
+
+ assert "crf-search" in crf_search_args
+ assert_push "transfer_started", %{status: "transfer_started", video_id: ^video_id}
+ end)
+ after
+ Application.delete_env(:reencodarr, :worker_token)
+ end
+
+ test "returns in-progress work when an assigned worker asks again" do
+ token = "test-worker-token"
+ Application.put_env(:reencodarr, :worker_token, token)
+
+ {:ok, video} = Fixtures.video_fixture(%{state: :analyzed})
+ video_id = video.id
+
+ assert {:ok, socket} = connect(WorkerSocket, %{"token" => token})
+ assert {:ok, _join_payload, socket} = subscribe_and_join(socket, "workers:crf_search")
+
+ assert_reply push(socket, "announce", announce_payload(worker_id: "worker-a")),
+ :ok,
+ %{accepted: true, protocol_version: 1}
+
+ assert_reply push(socket, "pull_work", %{}),
+ :ok,
+ %{status: "job_assigned", video_id: ^video_id, crf_search_args: assigned_args}
+
+ assert_reply push(socket, "crf_search_progress", %{
+ "video_id" => video_id,
+ "percent" => 12.5,
+ "filename" => Path.basename(video.path),
+ "fps" => 24.0,
+ "crf" => 28.0,
+ "sample_num" => 1,
+ "total_samples" => 8
+ }),
+ :ok,
+ %{accepted: true, event: "crf_search_progress"}
+
+ assert_reply push(socket, "pull_work", %{}),
+ :ok,
+ %{
+ status: "job_in_progress",
+ video_id: ^video_id,
+ crf_search_args: in_progress_args
+ }
+
+ assert assigned_args == in_progress_args
+ assert video.path in in_progress_args
+ after
+ Application.delete_env(:reencodarr, :worker_token)
+ end
+
+ test "resends the input when a reconnecting worker still has transfer progress" do
+ token = "test-worker-token"
+ Application.put_env(:reencodarr, :worker_token, token)
+
+ with_temp_file("abcdefgh", ".mkv", fn path ->
+ {:ok, video} =
+ Fixtures.video_fixture(%{
+ state: :analyzed,
+ path: path,
+ size: 8
+ })
+
+ video_id = video.id
+
+ assert {:ok, _session} =
+ WorkerSessions.register(%{
+ server_worker_id: "worker-server-1",
+ client_worker_id: "worker-a",
+ protocol_version: 1,
+ version: "0.10.0",
+ capabilities: %{"crf_search" => true}
+ })
+
+ assert {:ok, _session} =
+ WorkerSessions.assign_video("worker-server-1", video_id, :receiving_input)
+
+ assert {:ok, _session} =
+ WorkerSessions.set_transfer_progress("worker-server-1", %{
+ job_id: Integer.to_string(video_id),
+ transfer_id: Integer.to_string(video_id),
+ filename: Path.basename(video.path),
+ percent: 50.0,
+ bytes_sent: 4,
+ total_bytes: 8,
+ bytes_per_second: 256,
+ eta: 2,
+ chunk_index: 1,
+ total_chunks: 2
+ })
+
+ assert {:ok, socket2} = connect(WorkerSocket, %{"token" => token})
+ assert {:ok, _join_payload, socket2} = subscribe_and_join(socket2, "workers:crf_search")
+
+ assert_reply push(socket2, "announce", announce_payload(worker_id: "worker-a")),
+ :ok,
+ %{accepted: true, protocol_version: 1}
+
+ assert_reply push(socket2, "pull_work", %{}),
+ :ok,
+ %{
+ status: "job_assigned",
+ video_id: ^video_id,
+ chunk_size_bytes: chunk_size_bytes
+ }
+
+ assert_push "transfer_started", %{
+ status: "transfer_started",
+ video_id: ^video_id,
+ transfer_id: transfer_id,
+ chunk_size_bytes: ^chunk_size_bytes
+ }
+
+ assert transfer_id == Integer.to_string(video_id)
+ end)
+ after
+ Application.delete_env(:reencodarr, :worker_token)
+ end
+
+ test "reports transfer failure instead of silently waiting when resend input is unavailable" do
+ token = "test-worker-token"
+ Application.put_env(:reencodarr, :worker_token, token)
+
+ missing_path =
+ Path.join(System.tmp_dir!(), "missing-worker-input-#{System.unique_integer()}.mkv")
+
+ {:ok, video} =
+ Fixtures.video_fixture(%{
+ state: :crf_searching,
+ path: missing_path,
+ size: 8,
+ crf_search_worker_id: "worker-a"
+ })
+
+ video_id = video.id
+
+ assert {:ok, socket} = connect(WorkerSocket, %{"token" => token})
+ assert {:ok, _join_payload, socket} = subscribe_and_join(socket, "workers:crf_search")
+
+ assert_reply push(socket, "announce", announce_payload(worker_id: "worker-a")),
+ :ok,
+ %{accepted: true, protocol_version: 1}
+
+ assert_reply push(socket, "pull_work", %{"input_missing" => true}),
+ :ok,
+ %{status: "job_assigned", video_id: ^video_id}
+
+ assert_push "transfer_failed", %{
+ status: "transfer_failed",
+ video_id: ^video_id,
+ transfer_id: transfer_id
+ }
+
+ assert transfer_id == Integer.to_string(video_id)
+ refute_push "transfer_started", _, 50
+ after
+ Application.delete_env(:reencodarr, :worker_token)
+ end
+
+ test "reattaches dispatched work when the session state is gone" do
+ token = "test-worker-token"
+ Application.put_env(:reencodarr, :worker_token, token)
+
+ with_temp_file("abcdefgh", ".mkv", fn path ->
+ {:ok, video} =
+ Fixtures.video_fixture(%{
+ path: path,
+ size: 8,
+ state: :crf_searching,
+ crf_search_worker_id: "worker-dispatch-gone"
+ })
+
+ video_id = video.id
+
+ assert {:ok, socket} = connect(WorkerSocket, %{"token" => token})
+ assert {:ok, _join_payload, socket} = subscribe_and_join(socket, "workers:crf_search")
+
+ assert_reply push(
+ socket,
+ "announce",
+ announce_payload(worker_id: "worker-dispatch-gone")
+ ),
+ :ok,
+ %{accepted: true, protocol_version: 1}
+
+ assert_reply push(socket, "pull_work", %{}),
+ :ok,
+ %{status: "job_in_progress", video_id: ^video_id}
+
+ assert Media.get_video(video_id).state == :crf_searching
+ assert WorkerSessions.get(socket.assigns.worker_id).phase == :crf_searching
+ refute_push "transfer_started", _, 50
+ end)
+ after
+ Application.delete_env(:reencodarr, :worker_token)
+ end
+
+ test "streams assigned media in ordered chunks with integrity data" do
+ token = "test-worker-token"
+ previous_chunk_size = Application.get_env(:reencodarr, :worker_chunk_size_bytes)
+ Application.put_env(:reencodarr, :worker_token, token)
+ Application.put_env(:reencodarr, :worker_chunk_size_bytes, 4)
+
+ on_exit(fn ->
+ if is_nil(previous_chunk_size) do
+ Application.delete_env(:reencodarr, :worker_chunk_size_bytes)
+ else
+ Application.put_env(:reencodarr, :worker_chunk_size_bytes, previous_chunk_size)
+ end
+ end)
+
+ chunk_size = 4
+ content = "abcdefgh"
+ content_size = byte_size(content)
+ first_chunk = binary_part(content, 0, chunk_size)
+ second_chunk = binary_part(content, chunk_size, content_size - chunk_size)
+ first_crc32 = :erlang.crc32(first_chunk)
+ second_crc32 = :erlang.crc32(second_chunk)
+
+ with_temp_file(content, ".mkv", fn path ->
+ {:ok, video} =
+ Fixtures.video_fixture(%{
+ path: path,
+ size: content_size,
+ state: :analyzed
+ })
+
+ assert {:ok, socket} = connect(WorkerSocket, %{"token" => token})
+ assert {:ok, _join_payload, socket} = subscribe_and_join(socket, "workers:crf_search")
+
+ assert_reply push(socket, "announce", announce_payload(worker_id: "worker-a")),
+ :ok,
+ %{accepted: true, protocol_version: 1}
+
+ assert_reply push(socket, "pull_work", %{}),
+ :ok,
+ %{
+ status: "job_assigned",
+ video_id: assigned_video_id,
+ chunk_size_bytes: ^chunk_size
+ }
+
+ assert assigned_video_id == video.id
+
+ assert_push "transfer_started", %{
+ status: "transfer_started",
+ video_id: ^assigned_video_id,
+ transfer_id: transfer_id,
+ size_bytes: content_size,
+ chunk_size_bytes: ^chunk_size,
+ total_bytes: content_size,
+ total_chunks: 2
+ }
+
+ session = WorkerSessions.get(socket.assigns.worker_id)
+ assert session.active_video_id == assigned_video_id
+ assert session.phase == :receiving_input
+ assert session.transfer_progress.video_id == assigned_video_id
+ assert session.transfer_progress.percent == 0.0
+ assert session.transfer_progress.bytes_sent == 0
+ assert session.transfer_progress.total_bytes == content_size
+ assert session.transfer_progress.chunk_index == 0
+ assert session.transfer_progress.total_chunks == 2
+
+ assert_push "transfer_chunk", {:binary, first_frame}
+
+ assert {:ok,
+ %{
+ video_id: ^assigned_video_id,
+ transfer_id: ^transfer_id,
+ chunk_index: 0,
+ total_chunks: 2,
+ bytes_sent: ^chunk_size,
+ total_bytes: ^content_size,
+ crc32: ^first_crc32,
+ data: ^first_chunk
+ }} = WorkerProtocol.parse_transfer_chunk_frame(first_frame)
+
+ refute_push "transfer_chunk", _, 50
+
+ assert_reply push(socket, "transfer_progress", %{
+ "job_id" => transfer_id,
+ "transfer_id" => transfer_id,
+ "video_id" => assigned_video_id,
+ "percent" => 50.0,
+ "received_bytes" => chunk_size,
+ "expected_bytes" => content_size,
+ "chunk_index" => 0,
+ "total_chunks" => 2
+ }),
+ :ok,
+ %{accepted: true, event: "transfer_progress"}
+
+ assert_push "transfer_chunk", {:binary, second_frame}
+
+ assert {:ok,
+ %{
+ video_id: ^assigned_video_id,
+ transfer_id: ^transfer_id,
+ chunk_index: 1,
+ total_chunks: 2,
+ bytes_sent: ^content_size,
+ total_bytes: ^content_size,
+ crc32: ^second_crc32,
+ data: ^second_chunk
+ }} = WorkerProtocol.parse_transfer_chunk_frame(second_frame)
+
+ refute_push "transfer_complete", _, 50
+
+ assert_reply push(socket, "transfer_progress", %{
+ "job_id" => transfer_id,
+ "transfer_id" => transfer_id,
+ "video_id" => assigned_video_id,
+ "percent" => 100.0,
+ "received_bytes" => content_size,
+ "expected_bytes" => content_size,
+ "chunk_index" => 1,
+ "total_chunks" => 2
+ }),
+ :ok,
+ %{accepted: true, event: "transfer_progress"}
+
+ session = WorkerSessions.get(socket.assigns.worker_id)
+ assert session.active_video_id == assigned_video_id
+ assert session.phase == :input_ready
+ assert session.transfer_progress.video_id == assigned_video_id
+ assert session.transfer_progress.bytes_sent == content_size
+ assert session.transfer_progress.total_bytes == content_size
+ assert session.transfer_progress.chunk_index == 1
+ assert session.transfer_progress.total_chunks == 2
+ assert session.transfer_progress.percent == 100.0
+
+ assert_push "transfer_complete", %{
+ status: "transfer_complete",
+ video_id: ^assigned_video_id,
+ transfer_id: ^transfer_id,
+ total_bytes: ^content_size,
+ total_chunks: 2
+ }
+
+ session = WorkerSessions.get(socket.assigns.worker_id)
+ assert session.active_video_id == assigned_video_id
+ assert session.phase == :input_ready
+ assert session.transfer_progress.video_id == assigned_video_id
+ end)
+ after
+ Application.delete_env(:reencodarr, :worker_token)
+ end
+
+ test "submits structured CRF results and completes the job" do
+ token = "test-worker-token"
+ Application.put_env(:reencodarr, :worker_token, token)
+ Phoenix.PubSub.subscribe(Reencodarr.PubSub, Events.channel())
+
+ {:ok, video} = Fixtures.video_fixture(%{state: :analyzed})
+ video_id = video.id
+
+ assert {:ok, socket} = connect(WorkerSocket, %{"token" => token})
+ assert {:ok, _join_payload, socket} = subscribe_and_join(socket, "workers:crf_search")
+
+ assert_reply push(socket, "announce", announce_payload(worker_id: "worker-a")),
+ :ok,
+ %{accepted: true, protocol_version: 1}
+
+ assert_reply push(socket, "pull_work", %{}),
+ :ok,
+ %{status: "job_assigned", video_id: ^video_id}
+
+ assert_reply push(socket, "crf_search_result", %{
+ "video_id" => video_id,
+ "results" => [
+ %{
+ "crf" => 26,
+ "score" => 94.1,
+ "percent" => 93.0,
+ "size" => "600 MB",
+ "params" => ["--preset", "4"]
+ },
+ %{
+ "crf" => 28,
+ "score" => 96.4,
+ "percent" => 95.0,
+ "size" => "520 MB",
+ "params" => ["--preset", "4"],
+ "chosen" => true
+ }
+ ]
+ }),
+ :ok,
+ %{accepted: true, event: "crf_search_result"}
+
+ assert_receive {:crf_search_vmaf_result, %{video_id: ^video_id, crf: 26.0, score: 94.1}}
+ assert_receive {:crf_search_vmaf_result, %{video_id: ^video_id, crf: 28.0, score: 96.4}}
+
+ vmafs = Media.get_vmafs_for_video(video_id)
+ assert Enum.any?(vmafs, &(&1.crf == 26.0 and &1.score == 94.1))
+ assert Enum.any?(vmafs, &(&1.crf == 28.0 and &1.score == 96.4))
+ assert WorkerSessions.get(socket.assigns.worker_id).active_video_id == nil
+
+ assert_reply push(socket, "crf_search_completed", %{
+ "video_id" => video_id,
+ "result" => "ok",
+ "chosen_crf" => 28
+ }),
+ :ok,
+ %{accepted: true, event: "crf_search_completed"}
+
+ assert Media.get_video(video_id).state == :crf_searched
+ assert Media.get_video(video_id).chosen_vmaf_id != nil
+ assert WorkerSessions.get(socket.assigns.worker_id).active_video_id == nil
+
+ assert_reply push(socket, "pull_work", %{}),
+ :ok,
+ %{status: "no_work"}
+
+ assert_reply push(socket, "crf_search_result", %{
+ "video_id" => video_id,
+ "results" => [
+ %{
+ "crf" => 28,
+ "score" => 96.4,
+ "percent" => 95.0,
+ "chosen" => true
+ }
+ ]
+ }),
+ :ok,
+ %{accepted: true, event: "crf_search_result"}
+
+ assert WorkerSessions.get(socket.assigns.worker_id).active_video_id == nil
+ after
+ Application.delete_env(:reencodarr, :worker_token)
+ end
+
+ test "clears stale active work when a worker asks after the video completed elsewhere" do
+ token = "test-worker-token"
+ Application.put_env(:reencodarr, :worker_token, token)
+
+ {:ok, video} = Fixtures.video_fixture(%{state: :analyzed})
+ video_id = video.id
+
+ assert {:ok, socket} = connect(WorkerSocket, %{"token" => token})
+ assert {:ok, _join_payload, socket} = subscribe_and_join(socket, "workers:crf_search")
+
+ assert_reply push(socket, "announce", announce_payload(worker_id: "worker-a")),
+ :ok,
+ %{accepted: true, protocol_version: 1}
+
+ assert_reply push(socket, "pull_work", %{}),
+ :ok,
+ %{status: "job_assigned", video_id: ^video_id}
+
+ _vmaf = Fixtures.vmaf_fixture(%{video_id: video_id, crf: 28.0, score: 96.4})
+ assert {:ok, _} = Media.mark_vmaf_as_chosen(video_id, 28.0)
+ assert {:ok, _} = Media.mark_as_crf_searched(Media.get_video(video_id))
+
+ assert_reply push(socket, "pull_work", %{}),
+ :ok,
+ %{status: "no_work"}
+
+ assert WorkerSessions.get(socket.assigns.worker_id).active_video_id == nil
+ after
+ Application.delete_env(:reencodarr, :worker_token)
+ end
+
+ test "uses an already chosen VMAF when completion omits chosen_crf" do
+ token = "test-worker-token"
+ Application.put_env(:reencodarr, :worker_token, token)
+ Phoenix.PubSub.subscribe(Reencodarr.PubSub, Events.channel())
+
+ {:ok, video} = Fixtures.video_fixture(%{state: :analyzed})
+ video_id = video.id
+
+ assert {:ok, socket} = connect(WorkerSocket, %{"token" => token})
+ assert {:ok, _join_payload, socket} = subscribe_and_join(socket, "workers:crf_search")
+
+ assert_reply push(socket, "announce", announce_payload(worker_id: "worker-a")),
+ :ok,
+ %{accepted: true, protocol_version: 1}
+
+ assert_reply push(socket, "pull_work", %{}),
+ :ok,
+ %{status: "job_assigned", video_id: ^video_id}
+
+ assert_reply push(socket, "crf_search_result", %{
+ "video_id" => video_id,
+ "results" => [
+ %{
+ "crf" => 24,
+ "score" => 94.2,
+ "percent" => 90.0,
+ "chosen" => true
+ }
+ ]
+ }),
+ :ok,
+ %{accepted: true, event: "crf_search_result"}
+
+ assert_reply push(socket, "crf_search_completed", %{
+ "video_id" => video_id,
+ "result" => "ok"
+ }),
+ :ok,
+ %{accepted: true, event: "crf_search_completed"}
+
+ assert Media.get_video(video_id).state == :crf_searched
+ assert Media.get_video(video_id).chosen_vmaf_id != nil
+ assert WorkerSessions.get(socket.assigns.worker_id).active_video_id == nil
+ after
+ Application.delete_env(:reencodarr, :worker_token)
+ end
+
+ test "persists completion VMAF results before completing the job" do
+ token = "test-worker-token"
+ Application.put_env(:reencodarr, :worker_token, token)
+ Phoenix.PubSub.subscribe(Reencodarr.PubSub, Events.channel())
+
+ {:ok, video} = Fixtures.video_fixture(%{state: :analyzed})
+ video_id = video.id
+
+ assert {:ok, socket} = connect(WorkerSocket, %{"token" => token})
+ assert {:ok, _join_payload, socket} = subscribe_and_join(socket, "workers:crf_search")
+
+ assert_reply push(socket, "announce", announce_payload(worker_id: "worker-a")),
+ :ok,
+ %{accepted: true, protocol_version: 1}
+
+ assert_reply push(socket, "pull_work", %{}),
+ :ok,
+ %{status: "job_assigned", video_id: ^video_id}
+
+ assert_reply push(socket, "crf_search_completed", %{
+ "video_id" => video_id,
+ "result" => "ok",
+ "results" => [
+ %{
+ "crf" => 13,
+ "vmaf_score" => 90.74086,
+ "encode_percent" => 108.0,
+ "predicted_encode_size" => 5_230_000_000,
+ "predicted_encode_time_secs" => 5160
+ },
+ %{
+ "crf" => 12,
+ "vmaf_score" => 94.86,
+ "encode_percent" => 8.0,
+ "predicted_encode_size" => 263_590_000,
+ "predicted_encode_time_secs" => 1260,
+ "chosen" => true
+ }
+ ]
+ }),
+ :ok,
+ %{accepted: true, event: "crf_search_completed"}
+
+ assert_receive {:crf_search_vmaf_result, %{video_id: ^video_id, crf: 13.0, score: 90.74086}}
+ assert_receive {:crf_search_vmaf_result, %{video_id: ^video_id, crf: 12.0, score: 94.86}}
+
+ vmafs = Media.get_vmafs_for_video(video_id)
+ assert Enum.any?(vmafs, &(&1.crf == 13.0 and &1.score == 90.74086))
+ assert Enum.any?(vmafs, &(&1.crf == 12.0 and &1.score == 94.86))
+
+ video = Media.get_video(video_id)
+ assert video.state == :crf_searched
+ assert video.chosen_vmaf_id != nil
+ assert WorkerSessions.get(socket.assigns.worker_id).active_video_id == nil
+ after
+ Application.delete_env(:reencodarr, :worker_token)
+ end
+
+ test "fails a completed CRF search when nothing was chosen" do
+ token = "test-worker-token"
+ Application.put_env(:reencodarr, :worker_token, token)
+
+ {:ok, video} = Fixtures.video_fixture(%{state: :analyzed})
+ video_id = video.id
+
+ assert {:ok, socket} = connect(WorkerSocket, %{"token" => token})
+ assert {:ok, _join_payload, socket} = subscribe_and_join(socket, "workers:crf_search")
+
+ assert_reply push(socket, "announce", announce_payload(worker_id: "worker-a")),
+ :ok,
+ %{accepted: true, protocol_version: 1}
+
+ assert_reply push(socket, "pull_work", %{}),
+ :ok,
+ %{status: "job_assigned", video_id: ^video_id}
+
+ assert_reply push(socket, "crf_search_result", %{
+ "video_id" => video_id,
+ "results" => [
+ %{
+ "crf" => 24,
+ "score" => 94.2,
+ "percent" => 90.0
+ }
+ ]
+ }),
+ :ok,
+ %{accepted: true, event: "crf_search_result"}
+
+ assert_reply push(socket, "crf_search_completed", %{
+ "video_id" => video_id,
+ "result" => "ok"
+ }),
+ :error,
+ %{reason: "invalid_crf_search_result"}
+
+ assert Media.get_video(video_id).state == :failed
+ assert is_nil(Media.get_video(video_id).chosen_vmaf_id)
+
+ assert_reply push(socket, "pull_work", %{}),
+ :ok,
+ %{status: "no_work"}
+ after
+ Application.delete_env(:reencodarr, :worker_token)
+ end
+
+ test "accepts a CRF result for already dispatched work before the session is rebuilt" do
+ token = "test-worker-token"
+ Application.put_env(:reencodarr, :worker_token, token)
+ Phoenix.PubSub.subscribe(Reencodarr.PubSub, Events.channel())
+
+ {:ok, video} =
+ Fixtures.video_fixture(%{
+ state: :crf_searching,
+ crf_search_worker_id: "worker-a",
+ video_codecs: ["h264"],
+ audio_codecs: ["aac"]
+ })
+
+ video_id = video.id
+
+ assert {:ok, socket} = connect(WorkerSocket, %{"token" => token})
+ assert {:ok, _join_payload, socket} = subscribe_and_join(socket, "workers:crf_search")
+
+ assert_reply push(socket, "crf_search_result", %{
+ "job_id" => Integer.to_string(video_id),
+ "video_id" => video_id,
+ "source_name" => Path.basename(video.path),
+ "crf" => 31.0,
+ "vmaf_score" => 96.2,
+ "predicted_encode_size" => 123_456,
+ "encode_percent" => 42.5,
+ "predicted_encode_time_secs" => 87.5,
+ "from_cache" => false
+ }),
+ :ok,
+ %{accepted: true, event: "crf_search_result"}
+
+ assert_receive {:crf_search_vmaf_result, %{video_id: ^video_id, crf: 31.0, score: 96.2}}
+
+ assert Enum.any?(
+ Media.get_vmafs_for_video(video_id),
+ &(&1.crf == 31.0 and &1.score == 96.2)
+ )
+ after
+ Application.delete_env(:reencodarr, :worker_token)
+ end
+
+ test "records typed failures and cancels active work" do
+ token = "test-worker-token"
+ Application.put_env(:reencodarr, :worker_token, token)
+ Phoenix.PubSub.subscribe(Reencodarr.PubSub, Events.channel())
+
+ {:ok, video} = Fixtures.video_fixture(%{state: :analyzed})
+ video_id = video.id
+
+ assert {:ok, socket} = connect(WorkerSocket, %{"token" => token})
+ assert {:ok, _join_payload, socket} = subscribe_and_join(socket, "workers:crf_search")
+
+ assert_reply push(socket, "announce", announce_payload(worker_id: "worker-a")),
+ :ok,
+ %{accepted: true, protocol_version: 1}
+
+ assert_reply push(socket, "pull_work", %{}),
+ :ok,
+ %{status: "job_assigned", video_id: ^video_id}
+
+ assert_reply push(socket, "video_failed", %{
+ "video_id" => video_id,
+ "stage" => "crf_search",
+ "category" => "timeout",
+ "message" => "timed out",
+ "code" => "EXIT_137",
+ "context" => %{"node" => "worker@host"},
+ "retriable" => true,
+ "stderr_excerpt" => "ab-av1 timed out"
+ }),
+ :ok,
+ %{accepted: true, event: "video_failed"}
+
+ assert_receive {:video_failed,
+ %{
+ video_id: ^video_id,
+ stage: :crf_search,
+ category: :timeout,
+ message: "timed out"
+ }}
+
+ assert Media.get_video(video_id).state == :failed
+ assert [_failure] = Media.get_video_failures(video_id)
+
+ {:ok, cancelled_video} = Fixtures.video_fixture(%{state: :analyzed})
+ cancelled_video_id = cancelled_video.id
+
+ assert {:ok, cancel_socket} = connect(WorkerSocket, %{"token" => token})
+
+ assert {:ok, _join_payload, cancel_socket} =
+ subscribe_and_join(cancel_socket, "workers:crf_search")
+
+ assert_reply push(cancel_socket, "announce", announce_payload(worker_id: "worker-b")),
+ :ok,
+ %{accepted: true, protocol_version: 1}
+
+ assert_reply push(cancel_socket, "pull_work", %{}),
+ :ok,
+ %{status: "job_assigned", video_id: ^cancelled_video_id}
+
+ assert_reply push(cancel_socket, "crf_search_completed", %{
+ "video_id" => cancelled_video_id,
+ "result" => "cancelled"
+ }),
+ :ok,
+ %{accepted: true, event: "crf_search_completed"}
+
+ assert_receive {:crf_search_completed, %{video_id: ^cancelled_video_id, result: :cancelled}}
+
+ assert Media.get_video(cancelled_video_id).state == :analyzed
+ after
+ Application.delete_env(:reencodarr, :worker_token)
+ end
+
+ test "keeps worker-dispatched active work when the worker disconnects" do
+ token = "test-worker-token"
+ Application.put_env(:reencodarr, :worker_token, token)
+
+ {:ok, video} = Fixtures.video_fixture(%{state: :analyzed})
+
+ assert {:ok, socket} = connect(WorkerSocket, %{"token" => token})
+ assert {:ok, _join_payload, socket} = subscribe_and_join(socket, "workers:crf_search")
+
+ assert_reply push(socket, "announce", announce_payload(worker_id: "worker-a")),
+ :ok,
+ %{accepted: true, protocol_version: 1}
+
+ assert_reply push(socket, "pull_work", %{}),
+ :ok,
+ %{status: "job_assigned", video_id: assigned_video_id}
+
+ assert assigned_video_id == video.id
+ assert Media.get_video(video.id).state == :crf_searching
+
+ Process.unlink(socket.channel_pid)
+ assert :ok = close(socket)
+
+ refreshed = Media.get_video(video.id)
+ assert refreshed.state == :crf_searching
+ assert refreshed.crf_search_worker_id == "worker-a"
+ assert WorkerSessions.list() == []
+ after
+ Application.delete_env(:reencodarr, :worker_token)
+ end
+
+ test "rejects missing or invalid tokens" do
+ Application.put_env(:reencodarr, :worker_token, "test-worker-token")
+
+ assert :error = connect(WorkerSocket, %{})
+ assert :error = connect(WorkerSocket, %{"token" => "wrong"})
+ after
+ Application.delete_env(:reencodarr, :worker_token)
+ end
+
+ test "rejects worker connections when distributed workers are disabled" do
+ previous_enabled = Application.get_env(:reencodarr, :distributed_worker_enabled)
+ previous_token = Application.get_env(:reencodarr, :worker_token)
+
+ on_exit(fn ->
+ if is_nil(previous_enabled) do
+ Application.delete_env(:reencodarr, :distributed_worker_enabled)
+ else
+ Application.put_env(:reencodarr, :distributed_worker_enabled, previous_enabled)
+ end
+
+ if is_nil(previous_token) do
+ Application.delete_env(:reencodarr, :worker_token)
+ else
+ Application.put_env(:reencodarr, :worker_token, previous_token)
+ end
+ end)
+
+ Application.put_env(:reencodarr, :distributed_worker_enabled, false)
+ Application.put_env(:reencodarr, :worker_token, "test-worker-token")
+
+ assert :error = connect(WorkerSocket, %{"token" => "test-worker-token"})
+ end
+
+ test "rejects invalid topics without crashing" do
+ assert {:error, %{reason: "unauthorized"}} =
+ WorkerChannel.join("workers:other", %{}, %{assigns: %{worker_id: "worker-1"}})
+ after
+ Application.delete_env(:reencodarr, :worker_token)
+ end
+ end
+
+ defp announce_payload(overrides \\ []) do
+ override_map =
+ overrides
+ |> Enum.into(%{})
+ |> Map.new(fn
+ {key, value} when is_atom(key) -> {Atom.to_string(key), value}
+ pair -> pair
+ end)
+
+ Map.merge(
+ %{
+ "worker_id" => "abav1-dev",
+ "protocol_version" => 1,
+ "version" => "0.10.0",
+ "capabilities" => %{"crf_search" => true}
+ },
+ override_map
+ )
+ end
+end
diff --git a/test/reencodarr_web/controllers/worker_file_controller_test.exs b/test/reencodarr_web/controllers/worker_file_controller_test.exs
new file mode 100644
index 00000000..c72dae1f
--- /dev/null
+++ b/test/reencodarr_web/controllers/worker_file_controller_test.exs
@@ -0,0 +1,49 @@
+defmodule ReencodarrWeb.WorkerFileControllerTest do
+ use ReencodarrWeb.ConnCase, async: false
+
+ test "requires worker bearer token", %{conn: conn} do
+ previous = Application.get_env(:reencodarr, :worker_token)
+ previous_transfer = Application.get_env(:reencodarr, :worker_transfer_token)
+ Application.put_env(:reencodarr, :worker_token, "test-token")
+ Application.delete_env(:reencodarr, :worker_transfer_token)
+
+ on_exit(fn ->
+ restore_env(:worker_token, previous)
+ restore_env(:worker_transfer_token, previous_transfer)
+ end)
+
+ conn = get(conn, ~p"/workers/files/123")
+
+ assert conn.status == 401
+ end
+
+ test "sends the video file to authorized workers", %{conn: conn} do
+ previous = Application.get_env(:reencodarr, :worker_token)
+ previous_transfer = Application.get_env(:reencodarr, :worker_transfer_token)
+ Application.put_env(:reencodarr, :worker_transfer_token, "transfer-token")
+
+ on_exit(fn ->
+ restore_env(:worker_token, previous)
+ restore_env(:worker_transfer_token, previous_transfer)
+ end)
+
+ path = Path.join(System.tmp_dir!(), "reencodarr-worker-file-#{System.unique_integer()}.mkv")
+ File.write!(path, "video bytes")
+ on_exit(fn -> File.rm(path) end)
+
+ {:ok, video} = Fixtures.video_fixture(%{path: path, size: byte_size("video bytes")})
+
+ conn =
+ conn
+ |> put_req_header("authorization", "Bearer transfer-token")
+ |> get(~p"/workers/files/#{video.id}")
+
+ assert conn.status == 200
+ assert conn.resp_body == "video bytes"
+ assert [content_type] = get_resp_header(conn, "content-type")
+ assert String.starts_with?(content_type, "application/octet-stream")
+ end
+
+ defp restore_env(key, nil), do: Application.delete_env(:reencodarr, key)
+ defp restore_env(key, value), do: Application.put_env(:reencodarr, key, value)
+end
diff --git a/test/reencodarr_web/live/dashboard_v2_live_test.exs b/test/reencodarr_web/live/dashboard_v2_live_test.exs
index 31daa867..19ccd0d5 100644
--- a/test/reencodarr_web/live/dashboard_v2_live_test.exs
+++ b/test/reencodarr_web/live/dashboard_v2_live_test.exs
@@ -7,7 +7,7 @@ defmodule ReencodarrWeb.DashboardLiveTest do
- Button interactions without internal state checking
- Event handling for service communication
"""
- use ReencodarrWeb.ConnCase, async: true
+ use ReencodarrWeb.ConnCase, async: false
import Phoenix.LiveViewTest
@@ -24,6 +24,7 @@ defmodule ReencodarrWeb.DashboardLiveTest do
assert html =~ "Media Library Sync"
assert html =~ "Sonarr"
assert html =~ "Radarr"
+ assert html =~ "Open workers"
assert html =~ ~s(id="dashboard-root")
assert html =~ ~s(phx-hook="DashboardAnimations")
assert html =~ ~s(id="dashboard-active-work")
@@ -82,6 +83,27 @@ defmodule ReencodarrWeb.DashboardLiveTest do
assert html =~ "initial-queue-preview.mkv"
end
+ test "shows the worker websocket token when configured", %{conn: conn} do
+ previous_token = Application.get_env(:reencodarr, :worker_token)
+ Application.put_env(:reencodarr, :worker_token, "deploy-test-worker-token")
+
+ on_exit(fn ->
+ if previous_token do
+ Application.put_env(:reencodarr, :worker_token, previous_token)
+ else
+ Application.delete_env(:reencodarr, :worker_token)
+ end
+ end)
+
+ {:ok, _view, html} = live(conn, ~p"/")
+ expected_fingerprint = worker_token_fingerprint("deploy-test-worker-token")
+
+ assert html =~ "Worker WebSocket"
+ assert html =~ "deploy-test-worker-token"
+ assert html =~ expected_fingerprint
+ assert html =~ "/workers/socket/websocket?token="
+ end
+
test "includes chart data in the initial html response", %{conn: conn} do
{:ok, video} =
Fixtures.video_fixture(%{
@@ -494,4 +516,11 @@ defmodule ReencodarrWeb.DashboardLiveTest do
assert html =~ "Unknown sync service"
end
end
+
+ defp worker_token_fingerprint(token) do
+ :crypto.hash(:sha256, token)
+ |> Base.encode16(case: :lower)
+ |> String.slice(0, 12)
+ |> then(&"sha256:#{&1}")
+ end
end
diff --git a/test/reencodarr_web/live/workers_live_test.exs b/test/reencodarr_web/live/workers_live_test.exs
new file mode 100644
index 00000000..2759047a
--- /dev/null
+++ b/test/reencodarr_web/live/workers_live_test.exs
@@ -0,0 +1,248 @@
+defmodule ReencodarrWeb.WorkersLiveTest do
+ use ReencodarrWeb.ConnCase, async: false
+
+ import Phoenix.LiveViewTest
+
+ alias Reencodarr.AbAv1.WorkerSessions
+ alias Reencodarr.Fixtures
+
+ setup do
+ WorkerSessions.reset()
+ :ok
+ end
+
+ test "renders an empty state when no workers are connected", %{conn: conn} do
+ {:ok, _view, html} = live(conn, ~p"/workers")
+
+ assert html =~ "Workers"
+ assert html =~ "No workers connected."
+ assert html =~ ~s(href="/workers")
+ end
+
+ test "shows the transfer panel when a restarted worker is receiving input", %{conn: conn} do
+ {:ok, _session} =
+ WorkerSessions.register(%{
+ server_worker_id: "worker-server-1",
+ client_worker_id: "worker-client-1",
+ protocol_version: 1,
+ version: "0.10.0",
+ capabilities: %{"crf_search" => true}
+ })
+
+ {:ok, video} = Fixtures.video_fixture(%{state: :analyzed})
+
+ assert {:ok, view, html} = live(conn, ~p"/workers")
+ assert html =~ "worker-client-1"
+ assert html =~ "Idle"
+ refute html =~ "Receiving Input"
+
+ assert {:ok, _session} =
+ WorkerSessions.assign_video("worker-server-1", video.id, :receiving_input)
+
+ assert {:ok, _session} =
+ WorkerSessions.touch("worker-server-1", %{
+ cpu_percent: 87.5,
+ memory_bytes: 1_073_741_824,
+ memory_total_bytes: 4_294_967_296,
+ disk_free_bytes: 536_870_912_000,
+ disk_total_bytes: 1_099_511_627_776
+ })
+
+ assert {:ok, _session} =
+ WorkerSessions.set_transfer_progress("worker-server-1", %{
+ job_id: "job-1",
+ transfer_id: "job-1",
+ filename: Path.basename(video.path),
+ percent: 25.5,
+ bytes_sent: 2_621_440,
+ total_bytes: 10_485_760,
+ bytes_per_second: 1_048_576,
+ eta: 15,
+ chunk_index: 2,
+ total_chunks: 8
+ })
+
+ send(view.pid, {:worker_sessions_updated, %{sessions: WorkerSessions.list()}})
+ html = render(view)
+ assert html =~ "Receiving input"
+ assert html =~ "Receiving Input"
+ refute html =~ "CRF Search"
+ assert html =~ Path.basename(video.path)
+ assert html =~ "Target: 95 VMAF"
+ assert html =~ "CPU 87.5%"
+ assert html =~ "Mem 1.0 GiB / 4.0 GiB"
+ assert html =~ "Disk 500.0 GiB free / 1.0 TiB"
+ assert html =~ "2.5 MiB / 10.0 MiB"
+ assert html =~ "Chunk 3 / 8"
+ assert html =~ "25.5%"
+ assert html =~ "ETA 15s"
+ refute html =~ "CRF 28.0 -> 95.4 VMAF"
+ end
+
+ test "shows HTTP transfer progress without chunk text", %{conn: conn} do
+ {:ok, _session} =
+ WorkerSessions.register(%{
+ server_worker_id: "worker-server-http",
+ client_worker_id: "worker-client-http",
+ protocol_version: 1,
+ version: "0.10.0",
+ capabilities: %{"crf_search" => true}
+ })
+
+ {:ok, video} = Fixtures.video_fixture(%{state: :analyzed})
+
+ assert {:ok, _session} =
+ WorkerSessions.assign_video("worker-server-http", video.id, :receiving_input)
+
+ assert {:ok, _session} =
+ WorkerSessions.set_transfer_progress("worker-server-http", %{
+ job_id: "job-http",
+ transfer_id: "job-http",
+ filename: Path.basename(video.path),
+ percent: 50.0,
+ bytes_sent: 117_440_512,
+ total_bytes: 234_881_024,
+ bytes_per_second: 122_683_392,
+ eta: 1,
+ chunk_index: 0,
+ total_chunks: 0
+ })
+
+ {:ok, view, _html} = live(conn, ~p"/workers")
+ send(view.pid, {:worker_sessions_updated, %{sessions: WorkerSessions.list()}})
+
+ html = render(view)
+ assert html =~ "Receiving Input"
+ assert html =~ "112.0 MiB / 224.0 MiB"
+ assert html =~ "117.0 MiB/s"
+ assert html =~ "ETA 1s"
+ refute html =~ "Chunk"
+ end
+
+ test "transfer progress wins over stale crf progress in the worker card", %{conn: conn} do
+ {:ok, _session} =
+ WorkerSessions.register(%{
+ server_worker_id: "worker-server-stale-crf",
+ client_worker_id: "worker-client-stale-crf",
+ protocol_version: 1,
+ version: "0.10.0",
+ capabilities: %{"crf_search" => true}
+ })
+
+ {:ok, video} = Fixtures.video_fixture(%{state: :crf_searching})
+
+ assert {:ok, _session} =
+ WorkerSessions.assign_video("worker-server-stale-crf", video.id)
+
+ assert {:ok, _session} =
+ WorkerSessions.set_crf_search_progress("worker-server-stale-crf", %{
+ video_id: video.id,
+ percent: 62.0,
+ fps: 12.5,
+ eta: 90,
+ crf: 28.0,
+ sample_num: 3,
+ total_samples: 8
+ })
+
+ assert {:ok, _session} =
+ WorkerSessions.set_transfer_progress("worker-server-stale-crf", %{
+ job_id: Integer.to_string(video.id),
+ transfer_id: Integer.to_string(video.id),
+ video_id: video.id,
+ filename: Path.basename(video.path),
+ percent: 10.0,
+ bytes_sent: 1_048_576,
+ total_bytes: 10_485_760
+ })
+
+ {:ok, view, _html} = live(conn, ~p"/workers")
+ send(view.pid, {:worker_sessions_updated, %{sessions: WorkerSessions.list()}})
+
+ html = render(view)
+ assert html =~ "Receiving Input"
+ refute html =~ "CRF Search"
+ refute html =~ "Sample 3/8 - CRF 28.0"
+ end
+
+ test "renders the crf search panel when only crf search is active", %{conn: conn} do
+ {:ok, _session} =
+ WorkerSessions.register(%{
+ server_worker_id: "worker-server-2",
+ client_worker_id: "worker-client-2",
+ protocol_version: 1,
+ version: "0.10.0",
+ capabilities: %{"crf_search" => true}
+ })
+
+ {:ok, video} = Fixtures.video_fixture(%{state: :analyzed})
+
+ assert {:ok, _session} = WorkerSessions.assign_video("worker-server-2", video.id)
+
+ assert {:ok, _session} =
+ WorkerSessions.set_crf_search_progress("worker-server-2", %{
+ video_id: video.id,
+ percent: 62.0,
+ fps: 12.5,
+ eta: 90,
+ crf: 28.0,
+ sample_num: 3,
+ total_samples: 8
+ })
+
+ Fixtures.vmaf_fixture(%{
+ video_id: video.id,
+ crf: 28.0,
+ score: 95.4,
+ percent: 93.0
+ })
+
+ {:ok, view, _html} = live(conn, ~p"/workers")
+ send(view.pid, {:worker_sessions_updated, %{sessions: WorkerSessions.list()}})
+
+ html = render(view)
+ assert html =~ "CRF Search"
+ assert html =~ "Sample 3/8 - CRF 28.0"
+ assert html =~ "CRF 28.0 -> 95.4 VMAF"
+ refute html =~ "Receiving Input"
+ end
+
+ test "shows input ready after input transfer finishes before CRF progress", %{conn: conn} do
+ {:ok, _session} =
+ WorkerSessions.register(%{
+ server_worker_id: "worker-server-3",
+ client_worker_id: "worker-client-3",
+ protocol_version: 1,
+ version: "0.10.0",
+ capabilities: %{"crf_search" => true}
+ })
+
+ {:ok, video} = Fixtures.video_fixture(%{state: :crf_searching})
+
+ assert {:ok, _session} =
+ WorkerSessions.assign_video("worker-server-3", video.id, :receiving_input)
+
+ assert {:ok, _session} =
+ WorkerSessions.set_transfer_progress("worker-server-3", %{
+ job_id: Integer.to_string(video.id),
+ video_id: video.id,
+ transfer_id: Integer.to_string(video.id),
+ filename: Path.basename(video.path),
+ percent: 100.0,
+ bytes_sent: video.size,
+ total_bytes: video.size
+ })
+
+ assert {:ok, _session} = WorkerSessions.finish_transfer("worker-server-3")
+
+ {:ok, view, _html} = live(conn, ~p"/workers")
+ send(view.pid, {:worker_sessions_updated, %{sessions: WorkerSessions.list()}})
+
+ html = render(view)
+ assert html =~ "worker-client-3"
+ assert html =~ "Input ready"
+ assert html =~ "Input Ready"
+ refute html =~ "CRF Search"
+ refute html =~ "-%"
+ end
+end
diff --git a/test/support/channel_case.ex b/test/support/channel_case.ex
new file mode 100644
index 00000000..ed2cc9dd
--- /dev/null
+++ b/test/support/channel_case.ex
@@ -0,0 +1,20 @@
+defmodule ReencodarrWeb.ChannelCase do
+ @moduledoc false
+
+ use ExUnit.CaseTemplate
+
+ using do
+ quote do
+ @endpoint ReencodarrWeb.Endpoint
+
+ import Phoenix.ChannelTest
+
+ import ReencodarrWeb.ChannelCase
+ end
+ end
+
+ setup tags do
+ Reencodarr.DataCase.setup_sandbox(tags)
+ :ok
+ end
+end