perf(playout): parallelize playlist validation with dedup and worker pool

backend/playout/src/playlist.go

@@ -11,9 +11,11 @@ import (
 	"os"
 	"os/exec"
 	"path/filepath"
+	"runtime"
 	"sort"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
@@ -238,16 +240,32 @@ func LoadPlaylist(playlistPath string) ([]VideoFile, error) {
 // ValidateVideo uses ffprobe to enforce codec, fps, and keyframe interval (ADR-001)
 func ValidateVideo(videoPath string) (VideoFile, error) {
 	v := VideoFile{Path: videoPath}
-	var errors []string
 
-	// Run ffprobe
-	cmd := exec.Command("ffprobe", "-v", "quiet", "-print_format", "json", "-show_format", "-show_streams", videoPath)
-	output, err := cmd.Output()
-	if err != nil {
-		v.Validation = "ffprobe failed: " + err.Error()
-		return v, fmt.Errorf("ffprobe: %w", err)
+	// Run the stream/format probe and the keyframe probe concurrently: they are
+	// independent, and the keyframe probe (-show_frames) is the expensive one.
+	var streamOut []byte
+	var streamErr error
+	var keyframeErr error
+	var wg sync.WaitGroup
+	wg.Add(2)
+	go func() {
+		defer wg.Done()
+		cmd := exec.Command("ffprobe", "-v", "quiet", "-print_format", "json", "-show_format", "-show_streams", videoPath)
+		streamOut, streamErr = cmd.Output()
+	}()
+	go func() {
+		defer wg.Done()
+		keyframeErr = validateKeyframeInterval(videoPath)
+	}()
+	wg.Wait()
+
+	if streamErr != nil {
+		v.Validation = "ffprobe failed: " + streamErr.Error()
+		return v, fmt.Errorf("ffprobe: %w", streamErr)
 	}
 
+	var errors []string
+
 	// Parse JSON
 	var probe struct {
 		Streams []struct {
@@ -265,7 +283,7 @@ func ValidateVideo(videoPath string) (VideoFile, error) {
 			Duration string `json:"duration"`
 		} `json:"format"`
 	}
-	if err := json.Unmarshal(output, &probe); err != nil {
+	if err := json.Unmarshal(streamOut, &probe); err != nil {
 		v.Validation = "ffprobe JSON parse failed: " + err.Error()
 		return v, fmt.Errorf("ffprobe json: %w", err)
 	}
@@ -326,8 +344,7 @@ func ValidateVideo(videoPath string) (VideoFile, error) {
 		return v, fmt.Errorf("validation failed: %s", v.Validation)
 	}
 
-	// Validate keyframe interval <=2s (60 frames at 30fps)
-	keyframeErr := validateKeyframeInterval(videoPath)
+	// Keyframe interval check (probe already completed concurrently above).
 	if keyframeErr != nil {
 		errors = append(errors, keyframeErr.Error())
 	}
@@ -421,21 +438,62 @@ func ValidateVideoStrict(videoPath string) map[string]any {
 	return result
 }
 
-// ValidatePlaylist runs normalization validation on all entries
+// validateWorkers returns the number of concurrent validation workers.
+func validateWorkers() int {
+	n := runtime.NumCPU()
+	if v := os.Getenv("PLAYOUT_VALIDATE_WORKERS"); v != "" {
+		if parsed, err := strconv.Atoi(v); err == nil && parsed > 0 {
+			n = parsed
+		}
+	}
+	return n
+}
+
+// ValidatePlaylist runs normalization validation on all entries in parallel,
+// deduplicating repeated paths so each unique file is probed once.
 func ValidatePlaylist(playlistPath string) ([]VideoFile, error) {
 	videos, err := LoadPlaylist(playlistPath)
 	if err != nil {
 		return nil, err
 	}
 
-	for i := range videos {
-		validated, err := ValidateVideo(videos[i].Path)
-		videos[i] = validated
-		if err != nil {
-			fmt.Printf("Warning: %s - %s\n", videos[i].Path, err)
+	// Collect unique paths in order of first appearance.
+	var order []string
+	seen := make(map[string]bool, len(videos))
+	for _, v := range videos {
+		if !seen[v.Path] {
+			seen[v.Path] = true
+			order = append(order, v.Path)
 		}
 	}
 
+	// Validate unique paths concurrently with a bounded worker pool.
+	results := make(map[string]VideoFile, len(order))
+	var mu sync.Mutex
+	var wg sync.WaitGroup
+	sem := make(chan struct{}, validateWorkers())
+	for _, path := range order {
+		wg.Add(1)
+		sem <- struct{}{}
+		go func(p string) {
+			defer wg.Done()
+			defer func() { <-sem }()
+			vf, err := ValidateVideo(p)
+			if err != nil {
+				fmt.Printf("Warning: %s - %s\n", p, err)
+			}
+			mu.Lock()
+			results[p] = vf
+			mu.Unlock()
+		}(path)
+	}
+	wg.Wait()
+
+	// Map results back onto the playlist (preserving order and duplicates).
+	for i := range videos {
+		videos[i] = results[videos[i].Path]
+	}
+
 	return videos, nil
 }
 
@@ -632,11 +690,24 @@ func main() {
 	initRunID()
 	fmt.Println("Run ID:", runID)
 
-	// Validate playlist before starting (ADR-001 normalization enforcement)
-	if _, err := ValidatePlaylist(playlistPath); err != nil {
+	// Validate playlist before starting (ADR-001 normalization enforcement).
+	// Runs once here; start-playout.sh no longer pre-validates separately.
+	videos, err := ValidatePlaylist(playlistPath)
+	if err != nil {
 		fmt.Println("ERROR: Playlist validation failed:", err)
 		os.Exit(1)
 	}
+	invalid := 0
+	for _, v := range videos {
+		if !v.Valid {
+			fmt.Printf("ERROR: invalid media: %s — %s\n", v.Path, v.Validation)
+			invalid++
+		}
+	}
+	if invalid > 0 {
+		fmt.Printf("ERROR: Playlist validation failed: %d invalid entries\n", invalid)
+		os.Exit(1)
+	}
 
 	// Initialize R2 client (skip if not configured — local HLS via nginx still works)
 	if err := initR2Client(); err != nil {

backend/playout/src/playlist_test.go

@@ -465,6 +465,54 @@ func TestValidatePlaylist(t *testing.T) {
 	t.Skip("requires ffprobe mock or test media fixtures")
 }
 
+func TestValidateWorkersConfig(t *testing.T) {
+	os.Unsetenv("PLAYOUT_VALIDATE_WORKERS")
+	if got := validateWorkers(); got < 1 {
+		t.Errorf("validateWorkers() = %d, want >= 1", got)
+	}
+
+	os.Setenv("PLAYOUT_VALIDATE_WORKERS", "3")
+	if got := validateWorkers(); got != 3 {
+		t.Errorf("validateWorkers() = %d, want 3", got)
+	}
+
+	os.Setenv("PLAYOUT_VALIDATE_WORKERS", "invalid")
+	if got := validateWorkers(); got < 1 {
+		t.Errorf("validateWorkers() = %d, want fallback >= 1", got)
+	}
+
+	os.Unsetenv("PLAYOUT_VALIDATE_WORKERS")
+}
+
+func TestValidatePlaylist_Dedup(t *testing.T) {
+	ensureFFProbeInPath(t)
+
+	valid := filepath.Join("..", "testdata", "valid_720p30.mp4")
+	playlistDir := t.TempDir()
+	p := filepath.Join(playlistDir, "playlist.txt")
+	// Same file listed three times to exercise dedup + order preservation.
+	content := fmt.Sprintf("file '%s'\nfile '%s'\nfile '%s'\n", valid, valid, valid)
+	if err := os.WriteFile(p, []byte(content), 0o644); err != nil {
+		t.Fatalf("write playlist: %v", err)
+	}
+
+	videos, err := ValidatePlaylist(p)
+	if err != nil {
+		t.Fatalf("ValidatePlaylist error: %v", err)
+	}
+	if len(videos) != 3 {
+		t.Fatalf("len = %d, want 3", len(videos))
+	}
+	for i, v := range videos {
+		if !v.Valid {
+			t.Errorf("entry %d invalid: %s", i, v.Validation)
+		}
+		if v.Path != valid {
+			t.Errorf("entry %d path = %q, want %q", i, v.Path, valid)
+		}
+	}
+}
+
 func TestSyncToR2(t *testing.T) {
 	prevClient := r2Client
 	prevBucket := bucket