feat: add automatic media normalization, R2 retention cleanup, and seamless playout loops

backend/playout/src/playlist.go

@@ -3,6 +3,8 @@ package main
 import (
 	"bufio"
 	"context"
+	"crypto/sha256"
+	"encoding/hex"
 	"encoding/json"
 	"fmt"
 	"net/http"
@@ -16,6 +18,7 @@ import (
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/config"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/aws/aws-sdk-go-v2/service/s3/types"
 	"github.com/fsnotify/fsnotify"
 )
 
@@ -32,12 +35,20 @@ var (
 	putObject     = func(client *s3.Client, input *s3.PutObjectInput) (*s3.PutObjectOutput, error) {
 		return client.PutObject(context.TODO(), input)
 	}
-	bucket        = os.Getenv("R2_BUCKET_NAME")
-	playoutMode   = "starting" // "live", "slate", "degraded"
-	restartCount  = 0
-	restartWindow = time.Now()
-	slateDir      = "/app/slate"
-	hlsDir        = "/app/hls"
+	listObjects = func(client *s3.Client, input *s3.ListObjectsV2Input) (*s3.ListObjectsV2Output, error) {
+		return client.ListObjectsV2(context.TODO(), input)
+	}
+	deleteObjects = func(client *s3.Client, input *s3.DeleteObjectsInput) (*s3.DeleteObjectsOutput, error) {
+		return client.DeleteObjects(context.TODO(), input)
+	}
+	bucket             = os.Getenv("R2_BUCKET_NAME")
+	playoutMode        = "starting" // "live", "slate", "degraded"
+	restartCount       = 0
+	restartWindow      = time.Now()
+	slateDir           = "/app/slate"
+	hlsDir             = "/app/hls"
+	hlsSegmentDuration = envWithDefault("HLS_SEGMENT_DURATION", "6")
+	hlsWindowSize      = envWithDefault("HLS_WINDOW_SIZE", "24")
 
 	// Unique segment naming (STREAMING.md contract: r-{runID}/seg-{seq}.m4s)
 	runID       string
@@ -51,6 +62,14 @@ var (
 	consecutiveFails    int
 	lastFFmpegStartTime time.Time
 	lastFFmpegExitError error
+
+	// R2 retention cleanup counter
+	r2CleanupObjectsTotal int64
+
+	// Continuous playout loop metrics
+	loopsTotal      int64
+	lastRunDuration time.Duration
+	lastRunClean    bool
 )
 
 const (
@@ -61,6 +80,14 @@ const (
 	manifestCacheControl = "public, max-age=1, s-maxage=2, must-revalidate"
 )
 
+// envWithDefault returns the value of the environment variable or the default if not set.
+func envWithDefault(key, defaultVal string) string {
+	if v := os.Getenv(key); v != "" {
+		return v
+	}
+	return defaultVal
+}
+
 // isR2Configured returns true when the required R2 environment variables are set.
 func isR2Configured() bool {
 	return os.Getenv("R2_ENDPOINT") != "" &&
@@ -568,8 +595,21 @@ func resetHLSOutput(hlsDir string) error {
 
 func main() {
 	playlistPath := "/app/playlist/playlist.txt"
-	if len(os.Args) > 1 {
-		playlistPath = os.Args[1]
+	validateOnly := false
+	for _, arg := range os.Args[1:] {
+		switch arg {
+		case "-validate":
+			validateOnly = true
+		default:
+			if !strings.HasPrefix(arg, "-") {
+				playlistPath = arg
+			}
+		}
+	}
+
+	if validateOnly {
+		runValidateOnly(playlistPath)
+		return
 	}
 
 	fmt.Println("=== Linear Void Playout Service Starting ===")
@@ -591,6 +631,11 @@ func main() {
 		fmt.Println("R2 sync disabled:", err)
 	}
 
+	// Start R2 segment retention cleanup (hourly + once at startup)
+	if r2Client != nil {
+		go runR2CleanupLoop()
+	}
+
 	// Start HTTP health check server
 	http.HandleFunc("/health", HealthHandler)
 	http.HandleFunc("/metrics", func(w http.ResponseWriter, r *http.Request) {
@@ -607,6 +652,15 @@ func main() {
 			modeVal = 2
 		}
 		fmt.Fprintf(w, "linearvoid_playout_mode %d\n", modeVal)
+		fmt.Fprintln(w, "# HELP linearvoid_playout_r2_cleanup_objects_total Total R2 objects deleted by retention cleanup.")
+		fmt.Fprintln(w, "# TYPE linearvoid_playout_r2_cleanup_objects_total counter")
+		fmt.Fprintf(w, "linearvoid_playout_r2_cleanup_objects_total %d\n", r2CleanupObjectsTotal)
+		fmt.Fprintln(w, "# HELP linearvoid_playout_loops_total Total clean playlist completions (loops).")
+		fmt.Fprintln(w, "# TYPE linearvoid_playout_loops_total counter")
+		fmt.Fprintf(w, "linearvoid_playout_loops_total %d\n", loopsTotal)
+		fmt.Fprintln(w, "# HELP linearvoid_playout_run_duration_seconds Duration of the last FFmpeg run in seconds.")
+		fmt.Fprintln(w, "# TYPE linearvoid_playout_run_duration_seconds gauge")
+		fmt.Fprintf(w, "linearvoid_playout_run_duration_seconds %f\n", lastRunDuration.Seconds())
 	})
 	go func() {
 		fmt.Println("Health check server listening on :8081/health")
@@ -625,10 +679,34 @@ func main() {
 		// New run ID on each restart for unique R2 segment keys (STREAMING.md)
 		initRunID()
 
-		if err := resetHLSOutput(hlsDir); err != nil {
-			fmt.Println("Failed to reset HLS output:", err)
-			time.Sleep(5 * time.Second)
-			continue
+		// Reset HLS output only when the playlist changed, we are recovering from a
+		// crash/slate, or this is the first run. Otherwise continue the segment
+		// sequence so the loop is seamless for viewers (no gap on clean loops).
+		fingerprint, fpErr := playlistFingerprint(playlistPath)
+		if fpErr != nil {
+			fmt.Println("Warning: cannot fingerprint playlist:", fpErr)
+			fingerprint = ""
+		}
+		playlistChanged := fingerprint != playlistRevision
+		shouldReset := playlistChanged || playoutMode != "live" || !lastRunClean
+
+		windowSize := 24
+		if ws, err := strconv.Atoi(hlsWindowSize); err == nil && ws > 0 {
+			windowSize = ws
+		}
+
+		startNumber := 0
+		if shouldReset {
+			if err := resetHLSOutput(hlsDir); err != nil {
+				fmt.Println("Failed to reset HLS output:", err)
+				time.Sleep(5 * time.Second)
+				continue
+			}
+			playlistRevision = fingerprint
+		} else {
+			if n, err := hlsNextStartNumber(hlsDir, windowSize); err == nil {
+				startNumber = n
+			}
 		}
 
 		// Start filesystem watcher for real-time R2 segment upload (STREAMING.md).
@@ -652,17 +730,20 @@ func main() {
 			"-r", "30", "-g", "60", "-keyint_min", "60", "-sc_threshold", "0",
 			"-c:a", "aac", "-ar", "48000", "-ac", "2", "-b:a", "128k",
 			"-f", "hls",
-			"-hls_time", "6",
-			"-hls_list_size", "24",
+			"-hls_time", hlsSegmentDuration,
+			"-hls_list_size", hlsWindowSize,
 			"-hls_segment_type", "fmp4",
 			"-hls_fmp4_init_filename", "init.mp4",
 			"-hls_flags", "delete_segments+independent_segments",
 			"-hls_segment_filename", filepath.Join(hlsDir, "seg-%06d.m4s"),
+			"-start_number", strconv.Itoa(startNumber),
 			filepath.Join(hlsDir, "playlist.m3u8"))
 
 		runErr := cmd.Run()
 		runDuration := time.Since(lastFFmpegStartTime)
 		lastFFmpegExitError = runErr
+		lastRunDuration = runDuration
+		lastRunClean = runErr == nil
 
 		if watchCancel != nil {
 			watchCancel() // Stop the filesystem watcher
@@ -685,6 +766,7 @@ func main() {
 		// Differentiate clean playlist completion from FFmpeg crash.
 		// A nil error means FFmpeg completed the playlist normally — not a crash.
 		if runErr == nil {
+			loopsTotal++
 			fmt.Printf("[%s] Playlist completed normally after %s. Looping...\n",
 				time.Now().UTC().Format(time.RFC3339), runDuration.Round(time.Second))
 			// If we were in slate mode, a successful run means recovery
@@ -747,4 +829,171 @@ func syncLatestToR2(hlsDir string) error {
 	fmt.Println("Synced init.mp4 to R2")
 
 	return nil
+}
+
+// r2SegmentRetentionHours returns the R2 segment retention window in hours.
+func r2SegmentRetentionHours() time.Duration {
+	h, err := strconv.Atoi(envWithDefault("R2_SEGMENT_RETENTION_HOURS", "24"))
+	if err != nil || h <= 0 {
+		h = 24
+	}
+	return time.Duration(h) * time.Hour
+}
+
+// cleanupOldSegments deletes R2 objects under the hls/ prefix that are older than
+// the retention window, excluding the live manifest and any non-run keys.
+// It returns the number of objects deleted.
+func cleanupOldSegments(ctx context.Context) (int, error) {
+	if r2Client == nil {
+		if err := initR2Client(); err != nil {
+			return 0, err
+		}
+	}
+
+	cutoff := time.Now().UTC().Add(-r2SegmentRetentionHours())
+
+	var stale []string
+	var token *string
+	for {
+		output, err := listObjects(r2Client, &s3.ListObjectsV2Input{
+			Bucket:            aws.String(bucket),
+			Prefix:            aws.String("hls/"),
+			ContinuationToken: token,
+		})
+		if err != nil {
+			return 0, err
+		}
+		for _, obj := range output.Contents {
+			key := aws.ToString(obj.Key)
+			if key == "hls/playlist.m3u8" || key == "hls/playlist.m3u8.tmp" {
+				continue
+			}
+			// Only consider segment/init objects under a run directory: hls/r-.../...
+			if !strings.HasPrefix(key, "hls/r-") {
+				continue
+			}
+			if obj.LastModified != nil && obj.LastModified.Before(cutoff) {
+				stale = append(stale, key)
+			}
+		}
+		if output.IsTruncated == nil || !*output.IsTruncated {
+			break
+		}
+		token = output.NextContinuationToken
+	}
+
+	if len(stale) == 0 {
+		return 0, nil
+	}
+
+	deleted := 0
+	for i := 0; i < len(stale); i += 1000 {
+		end := i + 1000
+		if end > len(stale) {
+			end = len(stale)
+		}
+		batch := stale[i:end]
+		objects := make([]types.ObjectIdentifier, len(batch))
+		for j, key := range batch {
+			objects[j] = types.ObjectIdentifier{Key: aws.String(key)}
+		}
+		if _, err := deleteObjects(r2Client, &s3.DeleteObjectsInput{
+			Bucket: aws.String(bucket),
+			Delete: &types.Delete{Objects: objects, Quiet: aws.Bool(true)},
+		}); err != nil {
+			return deleted, err
+		}
+		for _, key := range batch {
+			fmt.Printf("R2 cleanup: deleted %s\n", key)
+		}
+		deleted += len(batch)
+	}
+
+	r2CleanupObjectsTotal += int64(deleted)
+	return deleted, nil
+}
+
+// runR2CleanupLoop runs retention cleanup once at startup, then hourly.
+func runR2CleanupLoop() {
+	if _, err := cleanupOldSegments(context.Background()); err != nil {
+		fmt.Println("R2 cleanup error:", err)
+	}
+	ticker := time.NewTicker(time.Hour)
+	defer ticker.Stop()
+	for range ticker.C {
+		if _, err := cleanupOldSegments(context.Background()); err != nil {
+			fmt.Println("R2 cleanup error:", err)
+		}
+	}
+}
+
+// playlistFingerprint returns a short hash of the playlist file content, used to
+// detect playlist changes between loops.
+func playlistFingerprint(playlistPath string) (string, error) {
+	data, err := os.ReadFile(playlistPath)
+	if err != nil {
+		return "", err
+	}
+	sum := sha256.Sum256(data)
+	return hex.EncodeToString(sum[:]), nil
+}
+
+// hlsNextStartNumber returns the next segment sequence number to use, and prunes
+// stale segments outside the live window so they do not accumulate across loops.
+func hlsNextStartNumber(hlsDir string, windowSize int) (int, error) {
+	files, err := filepath.Glob(filepath.Join(hlsDir, "seg-*.m4s"))
+	if err != nil {
+		return 0, err
+	}
+
+	seqOf := func(f string) int {
+		base := filepath.Base(f)
+		num := strings.TrimSuffix(strings.TrimPrefix(base, "seg-"), ".m4s")
+		n, err := strconv.Atoi(num)
+		if err != nil {
+			return -1
+		}
+		return n
+	}
+
+	maxSeq := -1
+	for _, f := range files {
+		if n := seqOf(f); n > maxSeq {
+			maxSeq = n
+		}
+	}
+	next := maxSeq + 1
+
+	// Keep the live window; delete older segments no longer referenced.
+	threshold := next - windowSize
+	for _, f := range files {
+		if n := seqOf(f); n >= 0 && n < threshold {
+			if err := os.Remove(f); err != nil && !os.IsNotExist(err) {
+				return next, err
+			}
+		}
+	}
+	return next, nil
+}
+
+// runValidateOnly validates the playlist and exits 0 (valid) or 1 (invalid/missing).
+func runValidateOnly(playlistPath string) {
+	videos, err := ValidatePlaylist(playlistPath)
+	if err != nil {
+		fmt.Println("ERROR: Playlist validation failed:", err)
+		os.Exit(1)
+	}
+	valid := true
+	for _, v := range videos {
+		if v.Valid {
+			fmt.Printf("OK: %s (%.1fs)\n", v.Path, v.Duration)
+		} else {
+			valid = false
+			fmt.Printf("INVALID: %s — %s\n", v.Path, v.Validation)
+		}
+	}
+	if !valid {
+		os.Exit(1)
+	}
+	fmt.Println("Playlist valid.")
 }

backend/playout/src/playlist_test.go

@@ -27,6 +27,8 @@ package main
 // See backend/playout/testdata/*.txt.
 
 import (
+	"context"
+	"fmt"
 	"io"
 	"net/http"
 	"net/http/httptest"
@@ -39,6 +41,7 @@ import (
 
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/aws/aws-sdk-go-v2/service/s3/types"
 )
 
 func ensureFFProbeInPath(t *testing.T) {
@@ -265,7 +268,7 @@ func TestHealthHandler(t *testing.T) {
 		`"status":"healthy"`,
 		`"service":"playout"`,
 		`"mode":"live"`,
-		`"restarts":3`,
+		`"crash_count":3`,
 		`"run_id":"r-20260810T120000Z"`,
 		`"last_segment_seq"`,
 		`"consecutive_failures"`,
@@ -588,4 +591,146 @@ func TestSyncToR2_NoCacheHeadersForUnknownExt(t *testing.T) {
 	if gotContentType != "" {
 		t.Fatalf("ContentType = %q, want empty", gotContentType)
 	}
+}
+
+func TestR2SegmentRetentionHours(t *testing.T) {
+	os.Setenv("R2_SEGMENT_RETENTION_HOURS", "")
+	if got := r2SegmentRetentionHours(); got != 24*time.Hour {
+		t.Errorf("default = %v, want 24h", got)
+	}
+
+	os.Setenv("R2_SEGMENT_RETENTION_HOURS", "48")
+	if got := r2SegmentRetentionHours(); got != 48*time.Hour {
+		t.Errorf("override = %v, want 48h", got)
+	}
+
+	os.Setenv("R2_SEGMENT_RETENTION_HOURS", "invalid")
+	if got := r2SegmentRetentionHours(); got != 24*time.Hour {
+		t.Errorf("invalid = %v, want 24h fallback", got)
+	}
+
+	os.Unsetenv("R2_SEGMENT_RETENTION_HOURS")
+}
+
+func TestCleanupOldSegments(t *testing.T) {
+	prevClient := r2Client
+	prevBucket := bucket
+	prevList := listObjects
+	prevDelete := deleteObjects
+	prevTotal := r2CleanupObjectsTotal
+	defer func() {
+		r2Client = prevClient
+		bucket = prevBucket
+		listObjects = prevList
+		deleteObjects = prevDelete
+		r2CleanupObjectsTotal = prevTotal
+	}()
+
+	bucket = "linearvoid-media"
+	r2Client = &s3.Client{}
+
+	now := time.Now().UTC()
+	old := now.Add(-25 * time.Hour)
+	recent := now.Add(-1 * time.Hour)
+
+	listObjects = func(_ *s3.Client, input *s3.ListObjectsV2Input) (*s3.ListObjectsV2Output, error) {
+		return &s3.ListObjectsV2Output{
+			Contents: []types.Object{
+				{Key: aws.String("hls/r-old/seg-1.m4s"), LastModified: &old},
+				{Key: aws.String("hls/r-old/init.mp4"), LastModified: &old},
+				{Key: aws.String("hls/r-new/seg-2.m4s"), LastModified: &recent},
+				{Key: aws.String("hls/playlist.m3u8"), LastModified: &old},
+				{Key: aws.String("hls/playlist.m3u8.tmp"), LastModified: &old},
+			},
+			IsTruncated: aws.Bool(false),
+		}, nil
+	}
+
+	var deletedKeys []string
+	deleteObjects = func(_ *s3.Client, input *s3.DeleteObjectsInput) (*s3.DeleteObjectsOutput, error) {
+		for _, obj := range input.Delete.Objects {
+			deletedKeys = append(deletedKeys, aws.ToString(obj.Key))
+		}
+		return &s3.DeleteObjectsOutput{}, nil
+	}
+
+	n, err := cleanupOldSegments(context.Background())
+	if err != nil {
+		t.Fatalf("cleanupOldSegments error: %v", err)
+	}
+	if n != 2 {
+		t.Fatalf("deleted = %d, want 2", n)
+	}
+
+	joined := strings.Join(deletedKeys, ",")
+	if !strings.Contains(joined, "hls/r-old/seg-1.m4s") || !strings.Contains(joined, "hls/r-old/init.mp4") {
+		t.Fatalf("unexpected deleted keys: %v", deletedKeys)
+	}
+	if strings.Contains(joined, "playlist.m3u8") {
+		t.Fatalf("manifest must not be deleted: %v", deletedKeys)
+	}
+	if r2CleanupObjectsTotal != 2 {
+		t.Fatalf("r2CleanupObjectsTotal = %d, want 2", r2CleanupObjectsTotal)
+	}
+}
+
+func TestPlaylistFingerprint(t *testing.T) {
+	tmpDir := t.TempDir()
+	p := filepath.Join(tmpDir, "playlist.txt")
+	if err := os.WriteFile(p, []byte("file '/app/media/a.mp4'\n"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+
+	fp1, err := playlistFingerprint(p)
+	if err != nil {
+		t.Fatalf("playlistFingerprint error: %v", err)
+	}
+	fp2, err := playlistFingerprint(p)
+	if err != nil {
+		t.Fatalf("playlistFingerprint error: %v", err)
+	}
+	if fp1 != fp2 {
+		t.Errorf("fingerprint not stable: %q vs %q", fp1, fp2)
+	}
+
+	if err := os.WriteFile(p, []byte("file '/app/media/b.mp4'\n"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	fp3, err := playlistFingerprint(p)
+	if err != nil {
+		t.Fatalf("playlistFingerprint error: %v", err)
+	}
+	if fp1 == fp3 {
+		t.Errorf("fingerprint should change when playlist content changes")
+	}
+}
+
+func TestHLSNextStartNumber(t *testing.T) {
+	tmpDir := t.TempDir()
+	// Create segments 0..29 with a live window of 10.
+	for i := 0; i < 30; i++ {
+		if err := os.WriteFile(filepath.Join(tmpDir, fmt.Sprintf("seg-%06d.m4s", i)), []byte("x"), 0o644); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	next, err := hlsNextStartNumber(tmpDir, 10)
+	if err != nil {
+		t.Fatalf("hlsNextStartNumber error: %v", err)
+	}
+	if next != 30 {
+		t.Errorf("next = %d, want 30", next)
+	}
+
+	// Segments < 20 should be pruned; 20..29 kept (the live window).
+	for i := 0; i < 20; i++ {
+		if _, err := os.Stat(filepath.Join(tmpDir, fmt.Sprintf("seg-%06d.m4s", i))); !os.IsNotExist(err) {
+			t.Errorf("seg-%06d should have been pruned", i)
+		}
+	}
+	for i := 20; i < 30; i++ {
+		if _, err := os.Stat(filepath.Join(tmpDir, fmt.Sprintf("seg-%06d.m4s", i))); err != nil {
+			t.Errorf("seg-%06d should be kept: %v", i, err)
+		}
+	}
 }