feat(playout): replace time-based R2 retention with run-count and size caps

2026-08-15 18:59:26 +02:00
1 parent 06b277e9c1
commit 6950bba85d
@@ -11,6 +11,7 @@ import (
 	"os"
 	"os/exec"
 	"path/filepath"
+	"sort"
 	"strconv"
 	"strings"
 	"time"
@@ -66,6 +67,10 @@ var (
 	// R2 retention cleanup counter
 	r2CleanupObjectsTotal int64

+	// R2 retention gauges
+	r2RunsRetained  int64
+	r2BytesRetained int64
+
 	// Continuous playout loop metrics
 	loopsTotal      int64
 	lastRunDuration time.Duration
@@ -78,6 +83,10 @@ const (
 	// Cache-Control header values for R2 uploads.
 	segmentCacheControl  = "public, max-age=31536000, immutable"
 	manifestCacheControl = "public, max-age=1, s-maxage=2, must-revalidate"
+
+	// R2 retention policy.
+	defaultR2KeepRuns          = 2
+	defaultR2MaxRetentionBytes = int64(10 * 1024 * 1024 * 1024) // 10 GiB
 )

 // envWithDefault returns the value of the environment variable or the default if not set.
@@ -658,6 +667,12 @@ func main() {
 		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_r2_runs_retained Number of R2 run folders retained by cleanup.")
+		fmt.Fprintln(w, "# TYPE linearvoid_playout_r2_runs_retained gauge")
+		fmt.Fprintf(w, "linearvoid_playout_r2_runs_retained %d\n", r2RunsRetained)
+		fmt.Fprintln(w, "# HELP linearvoid_playout_r2_bytes_retained Total bytes retained in R2 run folders.")
+		fmt.Fprintln(w, "# TYPE linearvoid_playout_r2_bytes_retained gauge")
+		fmt.Fprintf(w, "linearvoid_playout_r2_bytes_retained %d\n", r2BytesRetained)
 		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)
@@ -682,6 +697,15 @@ func main() {
 		// New run ID on each restart for unique R2 segment keys (STREAMING.md)
 		initRunID()

+		// Remove run folders older than the retention policy (keep the last N runs,
+		// cap total bytes) now that the next run has started. This never touches the
+		// current or previous run folders.
+		if r2Client != nil {
+			if _, err := cleanupOldRuns(context.Background()); err != nil {
+				fmt.Println("R2 cleanup error:", err)
+			}
+		}
+
 		// 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).
@@ -834,68 +858,51 @@ func syncLatestToR2(hlsDir string) error {
 	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
+// r2KeepRuns returns how many of the most recent run folders to keep in R2.
+func r2KeepRuns() int {
+	n, err := strconv.Atoi(envWithDefault("R2_KEEP_RUNS", "2"))
+	if err != nil || n < 1 {
+		return defaultR2KeepRuns
 	}
-	return time.Duration(h) * time.Hour
+	return n
 }

-// 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
-		}
+// r2MaxRetentionBytes returns the maximum total bytes to retain across run folders.
+func r2MaxRetentionBytes() int64 {
+	n, err := strconv.ParseInt(envWithDefault("R2_MAX_RETENTION_BYTES", "10737418240"), 10, 64)
+	if err != nil || n <= 0 {
+		return defaultR2MaxRetentionBytes
 	}
+	return n
+}

-	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
+// r2RunFromKey extracts the run folder name ("r-...") from an R2 object key.
+func r2RunFromKey(key string) (string, bool) {
+	rest := strings.TrimPrefix(key, "hls/")
+	if rest == key || !strings.HasPrefix(rest, "r-") {
+		return "", false
 	}
-
-	if len(stale) == 0 {
-		return 0, nil
+	if idx := strings.IndexByte(rest, '/'); idx > 0 {
+		return rest[:idx], true
 	}
+	return "", false
+}

+// runObjects groups the R2 object keys and total size for a single run folder.
+type runObjects struct {
+	keys []string
+	size int64
+}
+
+// deleteR2Keys deletes the given R2 object keys in batches of up to 1000.
+func deleteR2Keys(keys []string) (int, error) {
 	deleted := 0
-	for i := 0; i < len(stale); i += 1000 {
+	for i := 0; i < len(keys); i += 1000 {
 		end := i + 1000
-		if end > len(stale) {
-			end = len(stale)
+		if end > len(keys) {
+			end = len(keys)
 		}
-		batch := stale[i:end]
+		batch := keys[i:end]
 		objects := make([]types.ObjectIdentifier, len(batch))
 		for j, key := range batch {
 			objects[j] = types.ObjectIdentifier{Key: aws.String(key)}
@@ -911,6 +918,118 @@ func cleanupOldSegments(ctx context.Context) (int, error) {
 		}
 		deleted += len(batch)
 	}
+	return deleted, nil
+}
+
+// cleanupOldRuns lists run-scoped objects under hls/, keeps the most recent
+// R2_KEEP_RUNS runs, and deletes older runs. It also enforces the
+// R2_MAX_RETENTION_BYTES cap by removing oldest retained runs first, always
+// keeping at least the newest run. The live manifest keys are never deleted.
+func cleanupOldRuns(ctx context.Context) (int, error) {
+	if r2Client == nil {
+		if err := initR2Client(); err != nil {
+			return 0, err
+		}
+	}
+
+	runs := map[string]*runObjects{}
+	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
+			}
+			run, ok := r2RunFromKey(key)
+			if !ok {
+				continue
+			}
+			r, exists := runs[run]
+			if !exists {
+				r = &runObjects{}
+				runs[run] = r
+			}
+			r.keys = append(r.keys, key)
+			if obj.Size != nil {
+				r.size += *obj.Size
+			}
+		}
+		if output.IsTruncated == nil || !*output.IsTruncated {
+			break
+		}
+		token = output.NextContinuationToken
+	}
+
+	if len(runs) == 0 {
+		r2RunsRetained = 0
+		r2BytesRetained = 0
+		return 0, nil
+	}
+
+	// Run IDs are ISO timestamps, so lexicographic order equals chronological order.
+	sortedRuns := make([]string, 0, len(runs))
+	for run := range runs {
+		sortedRuns = append(sortedRuns, run)
+	}
+	sort.Strings(sortedRuns) // ascending: oldest -> newest
+
+	deleteRun := map[string]bool{}
+	keepRuns := r2KeepRuns()
+
+	// Mark all but the newest keepRuns runs for deletion.
+	if len(sortedRuns) > keepRuns {
+		for _, run := range sortedRuns[:len(sortedRuns)-keepRuns] {
+			deleteRun[run] = true
+		}
+	}
+
+	// Enforce the total size cap on retained runs (oldest retained first), always
+	// keeping the newest run even if it alone exceeds the cap.
+	maxBytes := r2MaxRetentionBytes()
+	retained := make([]string, 0, keepRuns)
+	var retainedBytes int64
+	for _, run := range sortedRuns {
+		if !deleteRun[run] {
+			retained = append(retained, run)
+			retainedBytes += runs[run].size
+		}
+	}
+	for i := 0; retainedBytes > maxBytes && i < len(retained)-1; i++ {
+		run := retained[i]
+		deleteRun[run] = true
+		retainedBytes -= runs[run].size
+	}
+
+	var toDelete []string
+	for run, del := range deleteRun {
+		if del {
+			toDelete = append(toDelete, runs[run].keys...)
+		}
+	}
+
+	deleted, err := deleteR2Keys(toDelete)
+	if err != nil {
+		return deleted, err
+	}
+
+	keptRuns := int64(0)
+	keptBytes := int64(0)
+	for run, r := range runs {
+		if !deleteRun[run] {
+			keptRuns++
+			keptBytes += r.size
+		}
+	}
+	r2RunsRetained = keptRuns
+	r2BytesRetained = keptBytes

 	r2CleanupObjectsTotal += int64(deleted)
 	return deleted, nil
@@ -918,13 +1037,13 @@ func cleanupOldSegments(ctx context.Context) (int, error) {

 // runR2CleanupLoop runs retention cleanup once at startup, then hourly.
 func runR2CleanupLoop() {
-	if _, err := cleanupOldSegments(context.Background()); err != nil {
+	if _, err := cleanupOldRuns(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 {
+		if _, err := cleanupOldRuns(context.Background()); err != nil {
 			fmt.Println("R2 cleanup error:", err)
 		}
 	}