From f128f53ed5d12cc0940bc2a1bf9dc3668f8a39c2 Mon Sep 17 00:00:00 2001 From: David Date: Mon, 12 Oct 2026 14:17:57 +0200 Subject: [PATCH 1/3] Fix media broker connection budget for long-running commands --- .../encoder/sandbox/broker/main.go | 38 ++++- .../encoder/sandbox/broker/main_test.go | 143 ++++++++++++++++++ 2 files changed, 179 insertions(+), 2 deletions(-) diff --git a/deploy/encoding-booster/encoder/sandbox/broker/main.go b/deploy/encoding-booster/encoder/sandbox/broker/main.go index 55e02e5..4cf5d69 100644 --- a/deploy/encoding-booster/encoder/sandbox/broker/main.go +++ b/deploy/encoding-booster/encoder/sandbox/broker/main.go @@ -248,6 +248,21 @@ type connections struct { once sync.Once } +// Bound live sockets, not the number of seeks over a media command's lifetime. +const maxActiveConnections = 4096 + +type limitedConnection struct { + net.Conn + once sync.Once + release func() +} + +func (c *limitedConnection) Close() error { + err := c.Conn.Close() + c.once.Do(c.release) + return err +} + func (l *connections) Accept() (net.Conn, error) { select { case connection := <-l.queue: @@ -260,29 +275,48 @@ func (l *connections) Close() error { l.once.Do(func() { close(l.done) }); ret func (l *connections) Addr() net.Addr { return &net.UnixAddr{Name: "media-broker", Net: "unix"} } func serve(ctx context.Context, control *net.UnixConn, b *broker) { + serveWithConnectionLimit(ctx, control, b, maxActiveConnections) +} + +func serveWithConnectionLimit(ctx context.Context, control *net.UnixConn, b *broker, limit int) { + ctx, cancel := context.WithCancel(ctx) + defer cancel() defer control.Close() + slots := make(chan struct{}, limit) listener := &connections{queue: make(chan net.Conn), done: make(chan struct{})} server := &http.Server{Handler: b, ReadHeaderTimeout: 10 * time.Second, IdleTimeout: 30 * time.Second, MaxHeaderBytes: 8192, BaseContext: func(net.Listener) context.Context { return ctx }} go func() { <-ctx.Done(); control.Close(); server.Close(); listener.Close() }() go server.Serve(listener) - // Limit parser-created connections over the entire invocation to bound abuse. - for count := 0; count < 4096; count++ { + for { var request [2]byte n, err := control.Read(request[:]) if err != nil || n != 1 || request[0] != 'C' { return } + select { + case slots <- struct{}{}: + default: + // A response without a descriptor is a refused connection. Keep the + // control channel available so the parser can retry after closing one. + if _, _, err := control.WriteMsgUnix([]byte{'N'}, nil, nil); err != nil { + return + } + continue + } parent, child, err := socketPair(syscall.SOCK_STREAM) if err != nil { + <-slots return } connection, err := net.FileConn(parent) parent.Close() if err != nil { child.Close() + <-slots return } + connection = &limitedConnection{Conn: connection, release: func() { <-slots }} _, _, err = control.WriteMsgUnix([]byte{'C'}, syscall.UnixRights(int(child.Fd())), nil) child.Close() if err != nil { diff --git a/deploy/encoding-booster/encoder/sandbox/broker/main_test.go b/deploy/encoding-booster/encoder/sandbox/broker/main_test.go index 34091f7..b52db93 100644 --- a/deploy/encoding-booster/encoder/sandbox/broker/main_test.go +++ b/deploy/encoding-booster/encoder/sandbox/broker/main_test.go @@ -1,14 +1,157 @@ package main import ( + "bufio" "context" "io" + "net" "net/http" "net/http/httptest" + "os" "strings" + "syscall" "testing" + "time" ) +func startTestBroker(t *testing.T, limit int) (*net.UnixConn, context.CancelFunc) { + t.Helper() + parent, child, err := socketPair(syscall.SOCK_SEQPACKET) + if err != nil { + t.Fatal(err) + } + server, err := net.FileConn(parent) + parent.Close() + if err != nil { + child.Close() + t.Fatal(err) + } + client, err := net.FileConn(child) + child.Close() + if err != nil { + server.Close() + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + defer close(done) + serveWithConnectionLimit(ctx, server.(*net.UnixConn), newBroker(), limit) + }() + t.Cleanup(func() { + cancel() + client.Close() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Error("broker did not stop") + } + }) + return client.(*net.UnixConn), cancel +} + +func brokerConnection(t *testing.T, control *net.UnixConn) net.Conn { + t.Helper() + control.SetDeadline(time.Now().Add(5 * time.Second)) + if _, err := control.Write([]byte{'C'}); err != nil { + t.Fatal(err) + } + var reply [1]byte + oob := make([]byte, syscall.CmsgSpace(4)) + _, count, _, _, err := control.ReadMsgUnix(reply[:], oob) + if err != nil { + t.Fatal(err) + } + if count == 0 { + if reply[0] != 'N' { + t.Fatalf("unexpected refusal: %q", reply) + } + return nil + } + messages, err := syscall.ParseSocketControlMessage(oob[:count]) + if err != nil || len(messages) != 1 { + t.Fatalf("invalid descriptors: %v", err) + } + fds, err := syscall.ParseUnixRights(&messages[0]) + if err != nil || len(fds) != 1 { + t.Fatalf("invalid rights: %v", err) + } + file := os.NewFile(uintptr(fds[0]), "parser-connection") + connection, err := net.FileConn(file) + file.Close() + if err != nil { + t.Fatal(err) + } + connection.SetDeadline(time.Now().Add(5 * time.Second)) + return connection +} + +func TestBrokerAllowsMoreThanLifetimeBudgetOfSequentialConnections(t *testing.T) { + control, _ := startTestBroker(t, maxActiveConnections) + for i := 0; i < maxActiveConnections+100; i++ { + connection := brokerConnection(t, control) + if connection == nil { + t.Fatalf("connection %d refused", i+1) + } + // Invalid sources are rejected locally, without any outbound requests. + _, err := io.WriteString(connection, "GET /0 HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\n\r\n") + if err != nil { + t.Fatal(err) + } + response, err := io.ReadAll(connection) + connection.Close() + if err != nil || !strings.Contains(string(response), "403 Forbidden") { + t.Fatalf("request %d failed: %v", i+1, err) + } + } +} + +func TestBrokerBoundsLiveConnectionsAndReusesClosedSlots(t *testing.T) { + control, _ := startTestBroker(t, 2) + first := brokerConnection(t, control) + second := brokerConnection(t, control) + if first == nil || second == nil { + t.Fatal("initial connections refused") + } + defer second.Close() + if excess := brokerConnection(t, control); excess != nil { + excess.Close() + t.Fatal("exceeded live connection limit") + } + first.Close() + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + if replacement := brokerConnection(t, control); replacement != nil { + replacement.Close() + return + } + } + t.Fatal("closed connection slot was not released") +} + +func TestBrokerCancellationClosesLiveConnections(t *testing.T) { + control, cancel := startTestBroker(t, 2) + connection := brokerConnection(t, control) + if connection == nil { + t.Fatal("connection refused") + } + defer connection.Close() + // Get a response first to ensure the HTTP server has accepted the socket. + io.WriteString(connection, "GET /0 HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n") + response, err := http.ReadResponse(bufio.NewReader(connection), nil) + if err != nil { + t.Fatal(err) + } + io.Copy(io.Discard, response.Body) + response.Body.Close() + cancel() + var data [1]byte + _, err = connection.Read(data[:]) + if err != io.EOF { + t.Fatalf("expected closed connection, got %v", err) + } +} + func TestBrokerScopesCredentialsAndRanges(t *testing.T) { requests := 0 origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { From 5ac721a7865d629fffac5d16b53fe0c6396ee188 Mon Sep 17 00:00:00 2001 From: David Date: Mon, 12 Oct 2026 14:18:06 +0200 Subject: [PATCH 2/3] Fix publishing empty embeds after private playback --- lib/mave_core/playback/storage_policy.ex | 3 +- .../playback/storage_policy_test.exs | 71 ++++++++++++++++++- 2 files changed, 72 insertions(+), 2 deletions(-) diff --git a/lib/mave_core/playback/storage_policy.ex b/lib/mave_core/playback/storage_policy.ex index 6fbb3f3..09c127f 100644 --- a/lib/mave_core/playback/storage_policy.ex +++ b/lib/mave_core/playback/storage_policy.ex @@ -75,7 +75,8 @@ defmodule MaveCore.Playback.StoragePolicy do do: :ok, else: {:error, :playback_public_access_not_converged} else - {:error, :not_found} when visibility == :private -> + {:error, :not_found} -> + # Empty embeds have no manifest to verify in either visibility state. if is_nil(Repo.preload(embed, :asset).asset.current_video_id), do: :ok, else: {:error, :playback_storage_verification_failed} diff --git a/test/mave_core/playback/storage_policy_test.exs b/test/mave_core/playback/storage_policy_test.exs index 1459614..c18c43e 100644 --- a/test/mave_core/playback/storage_policy_test.exs +++ b/test/mave_core/playback/storage_policy_test.exs @@ -3,8 +3,10 @@ defmodule MaveCore.Playback.StoragePolicyTest do import Plug.Conn - alias MaveCore.{Accounts, Embeds} + alias MaveCore.{Accounts, Embeds, Playback} + alias MaveCore.Assets.Video alias MaveCore.Playback.Minio + alias MaveCore.Workers.PlaybackVisibilityWorker setup {Req.Test, :verify_on_exit!} @@ -59,4 +61,71 @@ defmodule MaveCore.Playback.StoragePolicyTest do assert_receive {:policy_resources, resources} refute Enum.any?(resources, &String.ends_with?(&1, "/#{private.hash}/*")) end + + test "empty embeds can transition from public to private and back", %{space: space} do + {:ok, embed} = Embeds.create_video_embed(space) + assert is_nil(embed.asset.current_video_id) + + for {visibility, pending} <- [private: :protecting, public: :publishing] do + expect_storage_sync(embed, visibility, 404) + + Oban.Testing.with_testing_mode(:manual, fn -> + assert {:ok, updated} = Playback.request_visibility(embed, visibility) + assert updated.playback_status == pending + end) + + assert :ok = PlaybackVisibilityWorker.perform(%Oban.Job{args: %{"embed_id" => embed.id}}) + updated = Repo.reload!(embed) + assert updated.playback_visibility == visibility + assert updated.playback_status == visibility + end + end + + test "a missing manifest still fails verification when a current video exists", %{space: space} do + {:ok, embed} = Embeds.create_video_embed(space) + video = %Video{} |> Video.changeset(%{asset_id: embed.asset_id}) |> Repo.insert!() + embed.asset |> change(current_video_id: video.id) |> Repo.update!() + embed = Repo.reload!(embed) |> Repo.preload(:space) + + for visibility <- [:private, :public] do + expect_storage_sync(embed, visibility, 404) + + assert {:error, :playback_storage_verification_failed} = + Minio.apply_visibility(embed, visibility) + end + end + + test "empty embeds do not ignore storage authorization failures", %{space: space} do + {:ok, embed} = Embeds.create_video_embed(space) + expect_storage_sync(embed, :public, 403) + + assert {:error, :playback_storage_verification_failed} = + Minio.apply_visibility(embed, :public) + end + + defp expect_storage_sync(embed, visibility, manifest_status) do + Req.Test.expect(__MODULE__, fn conn -> + assert conn.method == "PUT" + assert conn.query_string == "policy" + {:ok, body, conn} = read_body(conn) + resources = Jason.decode!(body)["Statement"] |> hd() |> Map.fetch!("Resource") + + assert Enum.any?(resources, &String.ends_with?(&1, "/#{embed.hash}/*")) == + (visibility == :public) + + send_resp(conn, 200, "") + end) + + Req.Test.expect(__MODULE__, fn conn -> + assert conn.method == "PUT" + assert conn.query_string == "cors" + send_resp(conn, 200, "") + end) + + Req.Test.expect(__MODULE__, if(manifest_status == 403, do: 4, else: 1), fn conn -> + assert conn.method == "HEAD" + assert String.ends_with?(conn.request_path, "/#{embed.hash}/manifest.json") + send_resp(conn, manifest_status, "") + end) + end end From a0a15007af5e15990412bd67b222bc8c355aa291 Mon Sep 17 00:00:00 2001 From: David Date: Mon, 12 Oct 2026 14:46:55 +0200 Subject: [PATCH 3/3] Resolve Elixir 1.20 compiler warnings and fix AIFF signatures --- lib/mave_core/accounts/user_notifier.ex | 2 - lib/mave_core/embeds.ex | 2 - lib/mave_core/embeds/manifest_publisher.ex | 2 - lib/mave_core/embeds/settings_serializer.ex | 2 - lib/mave_core/encoding_booster/hls_upload.ex | 1 - lib/mave_core/flow.ex | 6 --- .../flow/steps/ai_transcribe_audio_step.ex | 2 - .../flow/steps/ai_translate_subtitles_step.ex | 3 -- .../flow/steps/media_extract_frame_step.ex | 1 - .../steps/media_package_hls_audio_step.ex | 1 - .../steps/media_package_hls_variant_step.ex | 2 - .../flow/steps/media_transcode_audio_step.ex | 1 - .../steps/media_transcode_h264_ladder_step.ex | 42 ------------------- lib/mave_core/media/remote_source.ex | 2 +- lib/mave_core/media/signature.ex | 2 +- lib/mave_core/media/storage.ex | 22 ---------- lib/mave_core/public_api.ex | 2 - lib/mave_core/spaces.ex | 4 -- lib/mave_core/uploads.ex | 6 --- .../api/legacy_embed_controller.ex | 3 -- .../live/dashboard/videos/show.ex | 2 - .../plugs/event_ingest_rate_limit.ex | 4 -- test/mave_core/media/signature_test.exs | 11 +++++ 23 files changed, 13 insertions(+), 112 deletions(-) diff --git a/lib/mave_core/accounts/user_notifier.ex b/lib/mave_core/accounts/user_notifier.ex index cbbe5ce..b7ef512 100644 --- a/lib/mave_core/accounts/user_notifier.ex +++ b/lib/mave_core/accounts/user_notifier.ex @@ -240,8 +240,6 @@ defmodule MaveCore.Accounts.UserNotifier do end end - defp endpoint_http_port(_), do: nil - defp normalize_port(nil), do: nil defp normalize_port(port) when is_integer(port), do: port diff --git a/lib/mave_core/embeds.ex b/lib/mave_core/embeds.ex index a27b4bc..94eb515 100644 --- a/lib/mave_core/embeds.ex +++ b/lib/mave_core/embeds.ex @@ -1984,8 +1984,6 @@ defmodule MaveCore.Embeds do processing_player_published?(step_runs) end - defp processing_player_ready?(_run), do: false - defp processing_player_published?(step_runs) when is_list(step_runs) do Enum.any?(step_runs, fn %StepRun{ diff --git a/lib/mave_core/embeds/manifest_publisher.ex b/lib/mave_core/embeds/manifest_publisher.ex index dfbaaca..c2aa4b4 100644 --- a/lib/mave_core/embeds/manifest_publisher.ex +++ b/lib/mave_core/embeds/manifest_publisher.ex @@ -399,8 +399,6 @@ defmodule MaveCore.Embeds.ManifestPublisher do |> Enum.map(&stringify_keys/1) end - defp manifest_video_renditions(_, _hls_ready?), do: [] - defp playback_original_url( %Video{max_width: width, max_height: height}, renditions, diff --git a/lib/mave_core/embeds/settings_serializer.ex b/lib/mave_core/embeds/settings_serializer.ex index e3d5349..bd5be53 100644 --- a/lib/mave_core/embeds/settings_serializer.ex +++ b/lib/mave_core/embeds/settings_serializer.ex @@ -510,8 +510,6 @@ defmodule MaveCore.Embeds.SettingsSerializer do end end - defp upload_url(_), do: nil - defp build_upload_url(key) when is_binary(key) and key != "" do upload_config = Application.get_env(:mave_core, :upload, []) trimmed_key = String.trim_leading(key, "/") diff --git a/lib/mave_core/encoding_booster/hls_upload.ex b/lib/mave_core/encoding_booster/hls_upload.ex index e666193..6435203 100644 --- a/lib/mave_core/encoding_booster/hls_upload.ex +++ b/lib/mave_core/encoding_booster/hls_upload.ex @@ -42,7 +42,6 @@ defmodule MaveCore.EncodingBooster.HLSUpload do {:ok, uploads} else {:error, _reason} = error -> error - _other -> {:error, :invalid_hls_upload_token} end rescue _error -> {:error, :invalid_hls_upload_request} diff --git a/lib/mave_core/flow.ex b/lib/mave_core/flow.ex index acd0390..8ca364c 100644 --- a/lib/mave_core/flow.ex +++ b/lib/mave_core/flow.ex @@ -4353,8 +4353,6 @@ defmodule MaveCore.Flow do uploaded_original_output(step_runs) || public_upload_playback_output(input) end - defp processing_playback_output(_run, step_runs), do: uploaded_original_output(step_runs) - defp public_upload_playback_output(input) do case public_upload_source_url(input) do url when is_binary(url) -> %{"original_url" => url} @@ -4514,8 +4512,6 @@ defmodule MaveCore.Flow do |> Repo.all() end - defp list_step_runs(_flow_run_id), do: [] - defp maybe_sync_completed_media_metadata( flow_run_id, step_type, @@ -4629,8 +4625,6 @@ defmodule MaveCore.Flow do end end - defp inspect_has_audio(_step_runs), do: nil - defp uploaded_original_output(step_runs) when is_list(step_runs) do Enum.find_value(step_runs, fn %StepRun{ diff --git a/lib/mave_core/flow/steps/ai_transcribe_audio_step.ex b/lib/mave_core/flow/steps/ai_transcribe_audio_step.ex index 8fc95b1..db09f34 100644 --- a/lib/mave_core/flow/steps/ai_transcribe_audio_step.ex +++ b/lib/mave_core/flow/steps/ai_transcribe_audio_step.ex @@ -697,8 +697,6 @@ defmodule MaveCore.Flow.Steps.AiTranscribeAudioStep do end end - defp heuristic_language_from_text(_), do: nil - defp normalize_label(label, _language) when is_binary(label) and label != "" do String.trim(label) end diff --git a/lib/mave_core/flow/steps/ai_translate_subtitles_step.ex b/lib/mave_core/flow/steps/ai_translate_subtitles_step.ex index ba6bbc4..4d6445c 100644 --- a/lib/mave_core/flow/steps/ai_translate_subtitles_step.ex +++ b/lib/mave_core/flow/steps/ai_translate_subtitles_step.ex @@ -316,8 +316,6 @@ defmodule MaveCore.Flow.Steps.AiTranslateSubtitlesStep do {:ok, Jason.encode!(%{"text" => String.trim(text), "segments" => json_segments})} end - defp translation_to_json(_), do: {:error, :empty_translation_response} - defp segments_to_vtt(segments) do valid_segments = segments @@ -461,7 +459,6 @@ defmodule MaveCore.Flow.Steps.AiTranslateSubtitlesStep do defp source_language_from_output(_), do: nil defp detect_language_from_text(text) when is_binary(text), do: LanguageDetector.detect(text) - defp detect_language_from_text(_), do: nil defp same_base_language?(left, right) do base_language(left) == base_language(right) diff --git a/lib/mave_core/flow/steps/media_extract_frame_step.ex b/lib/mave_core/flow/steps/media_extract_frame_step.ex index dafbaff..51fad1d 100644 --- a/lib/mave_core/flow/steps/media_extract_frame_step.ex +++ b/lib/mave_core/flow/steps/media_extract_frame_step.ex @@ -412,7 +412,6 @@ defmodule MaveCore.Flow.Steps.MediaExtractFrameStep do defp maybe_put_booster_instance(metadata, _timing), do: metadata - defp encoding_booster_error_code(:encoding_booster_busy), do: "busy" defp encoding_booster_error_code({:encoding_booster_http_status, status}), do: "http_#{status}" defp encoding_booster_error_code(_reason), do: "request_failed" diff --git a/lib/mave_core/flow/steps/media_package_hls_audio_step.ex b/lib/mave_core/flow/steps/media_package_hls_audio_step.ex index bbca990..5e35a47 100644 --- a/lib/mave_core/flow/steps/media_package_hls_audio_step.ex +++ b/lib/mave_core/flow/steps/media_package_hls_audio_step.ex @@ -429,7 +429,6 @@ defmodule MaveCore.Flow.Steps.MediaPackageHlsAudioStep do end end - defp encoding_booster_error_code(:encoding_booster_busy), do: "busy" defp encoding_booster_error_code({:encoding_booster_http_status, status}), do: "http_#{status}" defp encoding_booster_error_code(_reason), do: "request_failed" diff --git a/lib/mave_core/flow/steps/media_package_hls_variant_step.ex b/lib/mave_core/flow/steps/media_package_hls_variant_step.ex index 1864e36..6befffc 100644 --- a/lib/mave_core/flow/steps/media_package_hls_variant_step.ex +++ b/lib/mave_core/flow/steps/media_package_hls_variant_step.ex @@ -963,8 +963,6 @@ defmodule MaveCore.Flow.Steps.MediaPackageHlsVariantStep do end end - defp select_variant_output(_output, _codec, _size), do: %{} - defp ok_transcode_video_output?(output, codec, size) do is_map(output) and output["step_type"] == "media.transcode_video" and output["status"] == "ok" and output_matches_variant?(output, codec, size) diff --git a/lib/mave_core/flow/steps/media_transcode_audio_step.ex b/lib/mave_core/flow/steps/media_transcode_audio_step.ex index 0f84a94..50745bd 100644 --- a/lib/mave_core/flow/steps/media_transcode_audio_step.ex +++ b/lib/mave_core/flow/steps/media_transcode_audio_step.ex @@ -498,7 +498,6 @@ defmodule MaveCore.Flow.Steps.MediaTranscodeAudioStep do defp maybe_put_fallback(metadata, _key, nil), do: metadata defp maybe_put_fallback(metadata, key, value), do: Map.put(metadata, key, value) - defp encoding_booster_error_code(:encoding_booster_busy), do: "busy" defp encoding_booster_error_code({:encoding_booster_http_status, status}), do: "http_#{status}" defp encoding_booster_error_code({:audio_upload_verification_failed, _reason}), diff --git a/lib/mave_core/flow/steps/media_transcode_h264_ladder_step.ex b/lib/mave_core/flow/steps/media_transcode_h264_ladder_step.ex index 52a20ea..d164a7a 100644 --- a/lib/mave_core/flow/steps/media_transcode_h264_ladder_step.ex +++ b/lib/mave_core/flow/steps/media_transcode_h264_ladder_step.ex @@ -714,8 +714,6 @@ defmodule MaveCore.Flow.Steps.MediaTranscodeH264LadderStep do result end - defp maybe_report_encoding_booster_completed(result, _context, _variants, _timings), do: result - defp maybe_finalize_encoding_booster_variants( _context, _variants, @@ -727,41 +725,6 @@ defmodule MaveCore.Flow.Steps.MediaTranscodeH264LadderStep do {:ok, uploaded_variants, "gpu_encoding_booster", booster_result.metadata} end - defp maybe_finalize_encoding_booster_variants( - %{encoding_booster_dispatch: :direct} = context, - variants, - output_paths, - uploaded_variants, - booster_result, - :cpu - ) do - {hls_outputs, hls_errors} = - prepackage_hls_variants(context, variants, output_paths, true) - - metadata = - put_hls_prepackage_metadata(booster_result.metadata, hls_outputs, hls_errors) - - {:ok, attach_hls_outputs(uploaded_variants, hls_outputs), "encoding_booster", metadata} - end - - defp maybe_finalize_encoding_booster_variants( - context, - variants, - output_paths, - uploaded_variants, - booster_result, - :cpu - ) do - finalize_transcoded_variants( - context, - variants, - output_paths, - uploaded_variants, - "encoding_booster", - booster_result.metadata - ) - end - defp encode_booster_variants(context, variants, input_url, output_paths, profile) do variants |> Enum.reduce_while({:ok, []}, fn variant, {:ok, timings} -> @@ -906,8 +869,6 @@ defmodule MaveCore.Flow.Steps.MediaTranscodeH264LadderStep do |> BoosterChunking.chunked?() end - defp chunked_booster_source?(_context, _profile), do: false - defp maybe_add_encoding_booster_referer(options, input_url) do if Storage.ffmpeg_storage_url?(input_url) do case Storage.ffmpeg_input_referer() do @@ -2289,9 +2250,6 @@ defmodule MaveCore.Flow.Steps.MediaTranscodeH264LadderStep do defp normalize_string(value) when is_atom(value), do: value |> Atom.to_string() |> normalize_string() - defp normalize_string(value) when is_boolean(value), - do: value |> to_string() |> normalize_string() - defp normalize_string(_value), do: nil defp ffmpeg_progress(%{progress_reporter: nil}, _variants), do: nil diff --git a/lib/mave_core/media/remote_source.ex b/lib/mave_core/media/remote_source.ex index 92595d4..ff07146 100644 --- a/lib/mave_core/media/remote_source.ex +++ b/lib/mave_core/media/remote_source.ex @@ -96,7 +96,7 @@ defmodule MaveCore.Media.RemoteSource do end defp write_chunk(file, data, received, request, response) do - case IO.binwrite(file, data) do + case :file.write(file, data) do :ok -> response = Req.Response.put_private(response, @bytes_private_key, received + byte_size(data)) diff --git a/lib/mave_core/media/signature.ex b/lib/mave_core/media/signature.ex index 4b865f3..f7c31b3 100644 --- a/lib/mave_core/media/signature.ex +++ b/lib/mave_core/media/signature.ex @@ -100,7 +100,7 @@ defmodule MaveCore.Media.Signature do defp audio?(<<"fLaC", _rest::binary>>), do: true defp audio?(<<"RIFF", _size::binary-size(4), "WAVE", _rest::binary>>), do: true - defp audio?(<<"FORM", _size::binary-size(4), form, _rest::binary>>) + defp audio?(<<"FORM", _size::binary-size(4), form::binary-size(4), _rest::binary>>) when form in ["AIFF", "AIFC"], do: true defp audio?(<<"#!AMR", _rest::binary>>), do: true diff --git a/lib/mave_core/media/storage.ex b/lib/mave_core/media/storage.ex index 12a3890..48db20f 100644 --- a/lib/mave_core/media/storage.ex +++ b/lib/mave_core/media/storage.ex @@ -2026,17 +2026,6 @@ defmodule MaveCore.Media.Storage do ) {:error, {:server_side_multipart_copy_failed, reason}} - - other -> - _ = - abort_multipart_upload( - context.destination_bucket, - context.destination_path, - context.destination_storage_profile, - upload_id - ) - - {:error, {:server_side_multipart_copy_failed, other}} end {:error, reason} -> @@ -2167,17 +2156,6 @@ defmodule MaveCore.Media.Storage do ) {:error, {:multipart_copy_failed, reason}} - - other -> - _ = - abort_multipart_upload( - context.destination_bucket, - context.destination_path, - context.destination_storage_profile, - upload_id - ) - - {:error, {:multipart_copy_failed, other}} end {:error, reason} -> diff --git a/lib/mave_core/public_api.ex b/lib/mave_core/public_api.ex index abbc9ca..adc4d17 100644 --- a/lib/mave_core/public_api.ex +++ b/lib/mave_core/public_api.ex @@ -533,8 +533,6 @@ defmodule MaveCore.PublicApi do end end - defp current_video(_embed), do: nil - defp embed_payload(%Space{} = space, %Embed{type: :video} = embed) do current_video = current_video(embed) version = current_video_version(embed) diff --git a/lib/mave_core/spaces.ex b/lib/mave_core/spaces.ex index 5cf5c9b..d146ba5 100644 --- a/lib/mave_core/spaces.ex +++ b/lib/mave_core/spaces.ex @@ -785,8 +785,6 @@ defmodule MaveCore.Spaces do end end - defp jwt_space_candidates(_sub), do: [] - defp list_space_by_scope_id(id) do case list_space_by_id(id) do [] -> list_space_by_embed_id(id) @@ -1381,8 +1379,6 @@ defmodule MaveCore.Spaces do |> then(fn {headers, _size, _count, truncated?} -> {headers, truncated?} end) end - defp response_headers_to_map(_), do: {%{}, false} - defp reduce_response_header( _header, {headers, size, count, _truncated?}, diff --git a/lib/mave_core/uploads.ex b/lib/mave_core/uploads.ex index 6e2a025..5906860 100644 --- a/lib/mave_core/uploads.ex +++ b/lib/mave_core/uploads.ex @@ -238,8 +238,6 @@ defmodule MaveCore.Uploads do :ok end - defp maybe_warm_encoding_boosters(_scope, _metadata), do: :ok - defp encoding_booster_prewarmer do Application.get_env(:mave_core, :encoding_booster_prewarmer, EncodingBooster) end @@ -362,7 +360,6 @@ defmodule MaveCore.Uploads do audio_track_id: track.id }} else - nil -> {:error, :not_found} {:error, reason} -> {:error, reason} end end @@ -390,7 +387,6 @@ defmodule MaveCore.Uploads do subtitle_id: subtitle.id }} else - nil -> {:error, :not_found} {:error, reason} -> {:error, reason} end end @@ -817,8 +813,6 @@ defmodule MaveCore.Uploads do end end - defp parse_integer(value, _fallback) when is_integer(value), do: value - defp parse_integer(value, fallback) when is_binary(value) do case Integer.parse(value) do {parsed, ""} -> parsed diff --git a/lib/mave_core_web/controllers/api/legacy_embed_controller.ex b/lib/mave_core_web/controllers/api/legacy_embed_controller.ex index f19a228..709db1d 100644 --- a/lib/mave_core_web/controllers/api/legacy_embed_controller.ex +++ b/lib/mave_core_web/controllers/api/legacy_embed_controller.ex @@ -41,9 +41,6 @@ defmodule MaveCoreWeb.Api.LegacyEmbedController do {:error, error} when is_binary(error) -> conn |> put_status(:bad_request) |> json(%{error: error}) - - _error -> - conn |> put_status(:bad_request) |> json(%{error: @invalid_jwt_error}) end end diff --git a/lib/mave_core_web/live/dashboard/videos/show.ex b/lib/mave_core_web/live/dashboard/videos/show.ex index 1f9795f..b231e45 100644 --- a/lib/mave_core_web/live/dashboard/videos/show.ex +++ b/lib/mave_core_web/live/dashboard/videos/show.ex @@ -1164,8 +1164,6 @@ defmodule MaveCoreWeb.Dashboard.Videos.Show do {Map.put(video, :processing_player_ready, ready?), visible_after_ms} end - defp with_processing_player_readiness(video, _socket, _embed_id), do: {video, nil} - defp processing_player_visible_after_ms(socket, video, now_ms, ready_now?) do if socket.assigns[:video_processing] && is_map(socket.assigns[:video]) && diff --git a/lib/mave_core_web/plugs/event_ingest_rate_limit.ex b/lib/mave_core_web/plugs/event_ingest_rate_limit.ex index 6503a43..54b408d 100644 --- a/lib/mave_core_web/plugs/event_ingest_rate_limit.ex +++ b/lib/mave_core_web/plugs/event_ingest_rate_limit.ex @@ -159,8 +159,4 @@ defmodule MaveCoreWeb.Plugs.EventIngestRateLimit do ms = if timeout_ms > 0, do: timeout_ms, else: fallback_interval_ms max(div(ms + 999, 1000), 1) end - - defp retry_after_seconds(_timeout_ms, fallback_interval_ms) do - max(div(fallback_interval_ms + 999, 1000), 1) - end end diff --git a/test/mave_core/media/signature_test.exs b/test/mave_core/media/signature_test.exs index 86f713b..1587326 100644 --- a/test/mave_core/media/signature_test.exs +++ b/test/mave_core/media/signature_test.exs @@ -35,6 +35,17 @@ defmodule MaveCore.Media.SignatureTest do end end + test "accepts AIFF and AIFC form types and rejects truncated or unrelated forms" do + for form <- ["AIFF", "AIFC"] do + assert :ok = Signature.validate_audio(<<"FORM", 4::32, form::binary>>) + end + + for form <- ["", "A", "AIF", "ILBM", "AIFX"] do + assert {:error, :unsupported_audio_signature} = + Signature.validate_audio(<<"FORM", 4::32, form::binary>>) + end + end + test "accepts common video and audio signatures" do assert Signature.audio_or_video?(<<0, 0, 0, 24, "ftyp", "isom", 0::128>>) assert Signature.audio_or_video?(<<0x1A, 0x45, 0xDF, 0xA3, 0::128>>)