Fix media broker connections, empty embed publishing and compiler warnings

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 {

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) {

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
 

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{

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,

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, "/")

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}

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{

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

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)

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"
 

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"
 

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)

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}),