feat: close STREAMING.md production readiness gaps

- Added run ID + unique segment naming (r-{ISOTIME}/seg-{seq}.m4s) - Added fsnotify watcher for real-time R2 segment upload (~7s) - Added atomic manifest publication (temp key → CopyObject) - Fixed manifest Cache-Control: public, max-age=1, s-maxage=2, must-revalidate - Added hls.js error recovery: bounded retries (5x), exponential backoff, reconnecting UI - Enriched /health endpoint: run_id, active_item, last_segment_seq, last_r2_pub, consecutive_failures, unhealthy detection - Updated tests: HealthHandler (new fields + unhealthy), player error recovery test - Updated TODO.md, BACKLOG.md, DONE.md, MEMORY.md
2026-08-10 17:20:54 +02:00
1 parent f6ca89747e
commit 18459a9811
@@ -16,6 +16,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/fsnotify/fsnotify"
 )

 // VideoFile represents a normalized video entry in the playlist
@@ -37,14 +38,25 @@ var (
 	restartWindow = time.Now()
 	slateDir      = "/app/slate"
 	hlsDir        = "/app/hls"
+
+	// Unique segment naming (STREAMING.md contract: r-{runID}/seg-{seq}.m4s)
+	runID       string
+	segSequence uint64
+
+	// Health endpoint detail (STREAMING.md contract)
+	activeItem       string
+	playlistRevision string
+	lastSegmentSeq   uint64
+	lastR2PubTime    time.Time
+	consecutiveFails int
 )

 const (
 	maxRestartsPerMinute = 5
 	slateManifestName    = "playlist.m3u8"
 	// Cache-Control header values for R2 uploads.
-	segmentCacheControl = "public, max-age=31536000, immutable"
-	manifestCacheControl = "public, max-age=2"
+	segmentCacheControl  = "public, max-age=31536000, immutable"
+	manifestCacheControl = "public, max-age=1, s-maxage=2, must-revalidate"
 )

 // initR2Client initializes the R2 (S3-compatible) client
@@ -67,6 +79,84 @@ func initR2Client() error {
 	return nil
 }

+// initRunID generates a unique run/session ID for globally unique segment naming (STREAMING.md contract).
+func initRunID() {
+	runID = "r-" + time.Now().UTC().Format("20060102T150405Z")
+	segSequence = 0
+}
+
+// watchAndPublishSegments watches the HLS output directory for new .m4s segments
+// and uploads them to R2 in real time (STREAMING.md contract: per-segment R2 sync).
+func watchAndPublishSegments(ctx context.Context, hlsRunDir string) {
+	watcher, err := fsnotify.NewWatcher()
+	if err != nil {
+		fmt.Println("Warning: fsnotify unavailable, R2 sync falls back to batch:", err)
+		return
+	}
+	defer watcher.Close()
+
+	if err := watcher.Add(hlsRunDir); err != nil {
+		fmt.Println("Warning: cannot watch HLS dir:", err)
+		return
+	}
+
+	for {
+		select {
+		case event, ok := <-watcher.Events:
+			if !ok {
+				return
+			}
+			if event.Op&fsnotify.Create != 0 && strings.HasSuffix(event.Name, ".m4s") {
+				// Wait for FFmpeg to finish writing the segment
+				time.Sleep(500 * time.Millisecond)
+				segSequence++
+				lastSegmentSeq = segSequence
+				key := "live/" + runID + "/" + filepath.Base(event.Name)
+				if err := SyncToR2(event.Name, key); err != nil {
+					consecutiveFails++
+					fmt.Printf("Failed to sync segment %s: %v\n", event.Name, err)
+				} else {
+					consecutiveFails = 0
+					lastR2PubTime = time.Now()
+				}
+			}
+		case err, ok := <-watcher.Errors:
+			if !ok {
+				return
+			}
+			fmt.Println("fsnotify error:", err)
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+// publishManifest uploads the manifest atomically: temp key first, then copy to final key (STREAMING.md contract).
+func publishManifest(localManifestPath string) error {
+	if r2Client == nil {
+		if err := initR2Client(); err != nil {
+			return err
+		}
+	}
+
+	// 1. Upload to temp key
+	if err := SyncToR2(localManifestPath, "live/playlist.m3u8.tmp"); err != nil {
+		return fmt.Errorf("temp manifest upload failed: %w", err)
+	}
+
+	// 2. Atomic copy to final key
+	_, err := r2Client.CopyObject(context.TODO(), &s3.CopyObjectInput{
+		Bucket:     aws.String(bucket),
+		CopySource: aws.String(bucket + "/live/playlist.m3u8.tmp"),
+		Key:        aws.String("live/playlist.m3u8"),
+	})
+	if err != nil {
+		return fmt.Errorf("manifest atomic swap failed: %w", err)
+	}
+
+	return nil
+}
+
 // LoadPlaylist reads a playlist file and returns a list of VideoFile entries
 func LoadPlaylist(playlistPath string) ([]VideoFile, error) {
 	file, err := os.Open(playlistPath)
@@ -336,16 +426,32 @@ func SyncToR2(localPath, r2Key string) error {
 	return err
 }

-// HealthHandler provides a simple health check endpoint
+// HealthHandler provides a detailed health check endpoint (STREAMING.md contract)
 func HealthHandler(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Content-Type", "application/json")
 	status := "healthy"
-	if playoutMode == "slate" {
+	if playoutMode == "slate" || playoutMode == "degraded" {
 		status = "degraded"
 	}
-	w.WriteHeader(http.StatusOK)
-	fmt.Fprintf(w, `{"status":"%s","service":"playout","mode":"%s","restarts":%d,"timestamp":"%s"}`,
-		status, playoutMode, restartCount, time.Now().UTC().Format(time.RFC3339))
+	// Unhealthy if manifest has not advanced for >2 segment durations (12s)
+	if hlsDir != "" && time.Since(lastR2PubTime) > 12*time.Second && playoutMode != "starting" {
+		status = "unhealthy"
+	}
+
+	resp := map[string]any{
+		"status":              status,
+		"service":             "playout",
+		"mode":                playoutMode,
+		"restarts":            restartCount,
+		"run_id":              runID,
+		"active_item":         activeItem,
+		"playlist_revision":   playlistRevision,
+		"last_segment_seq":    lastSegmentSeq,
+		"last_r2_pub":         lastR2PubTime.UTC().Format(time.RFC3339),
+		"consecutive_failures": consecutiveFails,
+		"timestamp":           time.Now().UTC().Format(time.RFC3339),
+	}
+	json.NewEncoder(w).Encode(resp)
 }

 // activateSlate copies pre-rendered slate HLS into the live output directory.
@@ -450,6 +556,10 @@ func main() {
 	fmt.Println("Playlist:", playlistPath)
 	fmt.Println("R2 Bucket:", bucket)

+	// Initialize run ID for globally unique segment naming (STREAMING.md)
+	initRunID()
+	fmt.Println("Run ID:", runID)
+
 	// Validate playlist before starting (ADR-001 normalization enforcement)
 	if _, err := ValidatePlaylist(playlistPath); err != nil {
 		fmt.Println("ERROR: Playlist validation failed:", err)
@@ -493,12 +603,25 @@ func main() {
 			fmt.Printf("[%s] Slate active — attempting live playout recovery...\n", time.Now().Format(time.RFC3339))
 		}

+		// New run ID on each restart for unique segment paths (STREAMING.md)
+		initRunID()
+		runHlsDir := filepath.Join(hlsDir, runID)
+		if err := os.MkdirAll(runHlsDir, 0o755); err != nil {
+			fmt.Println("Failed to create run HLS dir:", err)
+			time.Sleep(5 * time.Second)
+			continue
+		}
+
 		if err := resetHLSOutput(hlsDir); err != nil {
 			fmt.Println("Failed to reset HLS output:", err)
 			time.Sleep(5 * time.Second)
 			continue
 		}

+		// Start filesystem watcher for real-time R2 segment upload (STREAMING.md)
+		watchCtx, watchCancel := context.WithCancel(context.Background())
+		go watchAndPublishSegments(watchCtx, runHlsDir)
+
 		fmt.Printf("[%s] Starting FFmpeg playout with browser-safe H.264/AAC HLS encoding...\n", time.Now().Format(time.RFC3339))

 		cmd := exec.Command("ffmpeg", "-re", "-f", "concat", "-safe", "0", "-i", playlistPath,
@@ -513,12 +636,12 @@ func main() {
 			"-hls_list_size", "24",
 			"-hls_segment_type", "fmp4",
 			"-hls_fmp4_init_filename", "init.mp4",
-			"-strftime", "1",
 			"-hls_flags", "delete_segments+independent_segments",
-			"-hls_segment_filename", "/app/hls/segment_%Y%m%d_%H%M%S.m4s",
-			"/app/hls/playlist.m3u8")
+			"-hls_segment_filename", filepath.Join(runHlsDir, "seg-%06d.m4s"),
+			filepath.Join(runHlsDir, "playlist.m3u8"))

 		runErr := cmd.Run()
+		watchCancel() // Stop the filesystem watcher

 		// Track restart and check for slate escalation
 		restartCount++
@@ -542,24 +665,30 @@ func main() {
 			deactivateSlate()
 		}

-		// After FFmpeg run, sync latest segments and manifest to R2
-		if err := syncLatestToR2(hlsDir); err != nil {
+		// After FFmpeg run, publish remaining segments and atomic manifest to R2
+		if err := syncLatestToR2(runHlsDir); err != nil {
 			fmt.Println("R2 sync warning:", err)
 		}

+		// Publish manifest atomically (STREAMING.md safe publication)
+		manifestPath := filepath.Join(runHlsDir, "playlist.m3u8")
+		if err := publishManifest(manifestPath); err != nil {
+			fmt.Println("Manifest publish warning:", err)
+		}
+
 		time.Sleep(5 * time.Second)
 	}
 }

 // syncLatestToR2 uploads new segments and the playlist manifest to R2
-func syncLatestToR2(hlsDir string) error {
-	files, err := filepath.Glob(filepath.Join(hlsDir, "segment_*.m4s"))
+func syncLatestToR2(hlsRunDir string) error {
+	files, err := filepath.Glob(filepath.Join(hlsRunDir, "seg-*.m4s"))
 	if err != nil {
 		return err
 	}

 	for _, f := range files {
-		key := "live/" + filepath.Base(f)
+		key := "live/" + runID + "/" + filepath.Base(f)
 		if err := SyncToR2(f, key); err != nil {
 			fmt.Printf("Failed to sync %s: %v\n", f, err)
 		} else {
@@ -567,16 +696,12 @@ func syncLatestToR2(hlsDir string) error {
 		}
 	}

-	// Sync the live playlist manifest (short TTL per ADR-001)
-	manifestPath := filepath.Join(hlsDir, "playlist.m3u8")
-	if err := SyncToR2(manifestPath, "live/playlist.m3u8"); err != nil {
+	// Sync init segment to run-specific path
+	initPath := filepath.Join(hlsRunDir, "init.mp4")
+	if err := SyncToR2(initPath, "live/"+runID+"/init.mp4"); err != nil {
 		return err
 	}
-	initPath := filepath.Join(hlsDir, "init.mp4")
-	if err := SyncToR2(initPath, "live/init.mp4"); err != nil {
-		return err
-	}
-	fmt.Println("Synced live playlist.m3u8 to R2")
+	fmt.Println("Synced init.mp4 to R2")

 	return nil
 }