dropped-frame tracke

cmd/mxl-player/main.go

@@ -467,7 +467,7 @@ func main() {
 		displayedVideoStride uint32 = placeholderStride
 
 		fps          float64
-		lastIndex    uint64
+		dropTracker  videoDropTracker
 		dropped      uint64
 		droppedTotal uint64
 		frameCount   uint64
@@ -530,6 +530,8 @@ func main() {
 			resized = false
 		}
 		var shownIndex uint64
+		var shownGeneration uint64
+		var shownSource playback.FeedConfig
 		hasFrame := false
 
 		frameCtx, frameCancel := context.WithTimeout(ctx, 100*time.Millisecond)
@@ -555,6 +557,8 @@ func main() {
 			}
 
 			shownIndex = pendingFrame.Frame.Index
+			shownGeneration = pendingFrame.Generation
+			shownSource = pendingFrame.Source
 			displayedVideoWidth = pendingFrame.Frame.Width
 			displayedVideoHeight = pendingFrame.Frame.Height
 			displayedVideoStride = pendingFrame.Frame.Stride
@@ -565,15 +569,14 @@ func main() {
 			panic(frameErr)
 		}
 
+		snapshot, hasSnapshot := player.Controller.Snapshot()
+
 		// stats
 		if hasFrame {
-			if lastIndex != 0 && shownIndex > lastIndex {
+			if gap := dropTracker.Observe(shownGeneration, shownSource, shownIndex); gap > 0 {
-				if g := shownIndex - lastIndex - 1; g > 0 {
+				dropped += gap
-					dropped += g
+				droppedTotal += gap
-					droppedTotal += g
-				}
 			}
-			lastIndex = shownIndex
 			frameCount++
 			if now := time.Now(); now.Sub(lastReport) >= time.Second {
 				dt := now.Sub(lastReport).Seconds()
@@ -588,7 +591,6 @@ func main() {
 		// end of stats
 		if r != nil {
 			gui.BeginFrame(time.Since(lastFrame), int32(r.Extent().Width), int32(r.Extent().Height))
-			snapshot, hasSnapshot := player.Controller.Snapshot()
 			if showStats {
 				cimgui.SetNextWindowPos(cimgui.Vec2{X: 0, Y: 0})
 				cimgui.SetNextWindowSize(cimgui.Vec2{X: 460, Y: 510})

cmd/mxl-player/playlist_file.go

@@ -11,8 +11,9 @@ import (
 )
 
 type playlistFile struct {
-	Entries []playlistFileEntry `json:"entries"`
+	Entries   []playlistFileEntry `json:"entries"`
-	Loop    bool                `json:"loop"`
+	Loop      bool                `json:"loop"`
+	OnFailure string              `json:"on_failure"`
 }
 
 type playlistFileEntry struct {
@@ -62,6 +63,16 @@ func decodePlaylistFile(reader io.Reader) (playback.Playlist, error) {
 		Entries: make([]playback.PlaylistEntry, len(file.Entries)),
 		Loop:    file.Loop,
 	}
+	switch file.OnFailure {
+	case "", "wait":
+		playlist.OnFailure = playback.PlaylistFailureWait
+	case "next":
+		playlist.OnFailure = playback.PlaylistFailureNext
+	default:
+		return playback.Playlist{}, fmt.Errorf(
+			"on_failure %q: %w", file.OnFailure, playback.ErrPlaylistFailurePolicy,
+		)
+	}
 	for index, entry := range file.Entries {
 		duration := time.Duration(0)
 		if entry.Duration != "" {

cmd/mxl-player/playlist_file_test.go

@@ -14,6 +14,7 @@ import (
 func TestDecodePlaylistFile(t *testing.T) {
 	input := `{
         "loop": true,
+		"on_failure": "next",
         "entries": [
             {
                 "name": "sync",
@@ -44,7 +45,8 @@ func TestDecodePlaylistFile(t *testing.T) {
 		t.Fatalf("decodePlaylistFile() error = %v", err)
 	}
 	want := playback.Playlist{
-		Loop: true,
+		Loop:      true,
+		OnFailure: playback.PlaylistFailureNext,
 		Entries: []playback.PlaylistEntry{
 			{
 				Name:          "sync",
@@ -69,7 +71,7 @@ func TestDecodePlaylistFile(t *testing.T) {
 			},
 		},
 	}
-	if len(got.Entries) != len(want.Entries) || got.Loop != want.Loop {
+	if len(got.Entries) != len(want.Entries) || got.Loop != want.Loop || got.OnFailure != want.OnFailure {
 		t.Fatalf("decodePlaylistFile() = %#v, want %#v", got, want)
 	}
 	for index := range want.Entries {
@@ -100,6 +102,7 @@ func TestDecodePlaylistFileRejectsInvalidInput(t *testing.T) {
 		{name: "malformed JSON", input: `{"entries": [`, wantText: "decode JSON"},
 		{name: "unknown field", input: `{"unknown": true}`, wantText: "unknown field"},
 		{name: "multiple roots", input: `{"entries": []} {"entries": []}`, wantText: "multiple root values"},
+		{name: "invalid failure policy", input: `{"on_failure":"skip","entries":[]}`, wantErr: playback.ErrPlaylistFailurePolicy},
 		{
 			name:     "invalid duration",
 			input:    `{"entries":[{"video":{"domain":"/video","uuid":"video"},"duration":"later"}]}`,

cmd/mxl-player/video_drop_tracker.go

@@ -0,0 +1,37 @@
+package main
+
+import "mxl-player/internal/playback"
+
+// videoDropTracker counts gaps only within one playback generation. Frame
+// indices belong to their source and cannot be compared across feed changes.
+type videoDropTracker struct {
+	generation uint64
+	source     playback.FeedConfig
+	lastIndex  uint64
+	hasIndex   bool
+}
+
+func (t *videoDropTracker) Observe(
+	generation uint64,
+	source playback.FeedConfig,
+	index uint64,
+) uint64 {
+	if !t.hasIndex ||
+		generation != t.generation ||
+		!sameVideoSource(source, t.source) ||
+		index <= t.lastIndex {
+		t.generation = generation
+		t.source = source
+		t.lastIndex = index
+		t.hasIndex = true
+		return 0
+	}
+
+	dropped := index - t.lastIndex - 1
+	t.lastIndex = index
+	return dropped
+}
+
+func sameVideoSource(a, b playback.FeedConfig) bool {
+	return a.Domain == b.Domain && a.UUID == b.UUID
+}

cmd/mxl-player/video_drop_tracker_test.go

@@ -0,0 +1,71 @@
+package main
+
+import (
+	"testing"
+
+	"mxl-player/internal/playback"
+)
+
+func TestVideoDropTracker(t *testing.T) {
+	tests := []struct {
+		name         string
+		observations [][2]uint64
+		want         []uint64
+	}{
+		{
+			name:         "counts gaps within generation",
+			observations: [][2]uint64{{1, 10}, {1, 11}, {1, 15}},
+			want:         []uint64{0, 0, 3},
+		},
+		{
+			name:         "higher index from new generation resets baseline",
+			observations: [][2]uint64{{1, 10}, {2, 1000000}, {2, 1000001}},
+			want:         []uint64{0, 0, 0},
+		},
+		{
+			name:         "lower index from new generation resets baseline",
+			observations: [][2]uint64{{1, 100}, {2, 5}, {2, 7}},
+			want:         []uint64{0, 0, 1},
+		},
+		{
+			name:         "index restart within generation resets baseline",
+			observations: [][2]uint64{{1, 100}, {1, 0}, {1, 1}},
+			want:         []uint64{0, 0, 0},
+		},
+		{
+			name:         "zero is a valid first index",
+			observations: [][2]uint64{{1, 0}, {1, 2}},
+			want:         []uint64{0, 1},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			var tracker videoDropTracker
+			source := playback.FeedConfig{Domain: "/mxl", UUID: "video"}
+			for index, observation := range test.observations {
+				got := tracker.Observe(observation[0], source, observation[1])
+				if got != test.want[index] {
+					t.Fatalf("Observe(%d, %d) = %d, want %d",
+						observation[0], observation[1], got, test.want[index])
+				}
+			}
+		})
+	}
+}
+
+func TestVideoDropTrackerResetsWhenSourceChanges(t *testing.T) {
+	var tracker videoDropTracker
+	first := playback.FeedConfig{Domain: "/mxl", UUID: "first"}
+	second := playback.FeedConfig{Domain: "/mxl", UUID: "second"}
+
+	if got := tracker.Observe(1, first, 10); got != 0 {
+		t.Fatalf("first Observe() = %d, want 0", got)
+	}
+	if got := tracker.Observe(1, second, 1000000); got != 0 {
+		t.Fatalf("source-changing Observe() = %d, want 0", got)
+	}
+	if got := tracker.Observe(1, second, 1000002); got != 1 {
+		t.Fatalf("same-source Observe() = %d, want 1", got)
+	}
+}

flow-def/video-4k.json

@@ -0,0 +1,41 @@
+{
+  "description": "sample for mxl reader go player",
+  "id": "5fbec3b1-1b0f-417d-9059-8b94a47197ed",
+  "tags": {
+    "urn:x-nmos:tag:grouphint/v1.0": [
+      "mxl-gst-testsrc pattern"
+    ]
+  },
+  "format": "urn:x-nmos:format:video",
+  "label": "SMPTE bars test video",
+  "parents": [],
+  "media_type": "video/v210",
+  "grain_rate": {
+    "numerator": 25,
+    "denominator": 1
+  },
+  "frame_width": 1920,
+  "frame_height": 1080,
+  "interlace_mode": "progressive",
+  "colorspace": "BT709",
+  "components": [
+    {
+      "name": "Y",
+      "width": 1920,
+      "height": 1080,
+      "bit_depth": 10
+    },
+    {
+      "name": "Cb",
+      "width": 960,
+      "height": 1080,
+      "bit_depth": 10
+    },
+    {
+      "name": "Cr",
+      "width": 960,
+      "height": 1080,
+      "bit_depth": 10
+    }
+  ]
+}

imgui.ini

@@ -1,30 +1,6 @@
 [Window][Debug##Default]
-Pos=519,181
-Size=400,398
-Collapsed=0
-
-[Window][Test]
 Pos=60,60
-Size=251,92
+Size=400,400
-Collapsed=0
-
-[Window][Stats]
-Size=460,510
-Collapsed=0
-
-[Window][Connection]
-Pos=42,264
-Size=605,416
-Collapsed=0
-
-[Window][Test slider]
-Pos=1370,0
-Size=550,1080
-Collapsed=0
-
-[Window][Settings]
-Pos=730,0
-Size=550,720
 Collapsed=0
 
 [Window][Settings & Info]
@@ -32,8 +8,7 @@ Pos=580,0
 Size=700,720
 Collapsed=0
 
-[Window][Seetings & Info]
+[Window][Stats]
-Pos=1220,0
+Size=460,510
-Size=700,1080
 Collapsed=0
 

internal/playback/playlist.go

@@ -11,8 +11,34 @@ var (
 	ErrPlaylistEntryEmpty        = errors.New("playlist entry must contain at least one feed")
 	ErrPlaylistSyncFeedsRequired = errors.New("synchronized playlist entry requires both video and audio feeds")
 	ErrPlaylistDurationNegative  = errors.New("playlist entry duration cannot be negative")
+	ErrPlaylistFailurePolicy     = errors.New("invalid playlist failure policy")
 )
 
+type PlaylistFailurePolicy uint8
+
+const (
+	PlaylistFailureWait PlaylistFailurePolicy = iota
+	PlaylistFailureNext
+)
+
+func (p PlaylistFailurePolicy) String() string {
+	switch p {
+	case PlaylistFailureWait:
+		return "wait"
+	case PlaylistFailureNext:
+		return "next"
+	default:
+		return fmt.Sprintf("PlaylistFailurePolicy(%d)", uint8(p))
+	}
+}
+
+func (p PlaylistFailurePolicy) Validate() error {
+	if p != PlaylistFailureWait && p != PlaylistFailureNext {
+		return ErrPlaylistFailurePolicy
+	}
+	return nil
+}
+
 type PlaylistFeed struct {
 	Domain string
 	UUID   string
@@ -27,8 +53,9 @@ type PlaylistEntry struct {
 }
 
 type Playlist struct {
-	Entries []PlaylistEntry
+	Entries   []PlaylistEntry
-	Loop    bool
+	Loop      bool
+	OnFailure PlaylistFailurePolicy
 }
 
 func (f PlaylistFeed) IsConfigured() bool {
@@ -65,6 +92,9 @@ func (e PlaylistEntry) Validate() error {
 }
 
 func (p Playlist) Validate() error {
+	if err := p.OnFailure.Validate(); err != nil {
+		return err
+	}
 	for index, entry := range p.Entries {
 		if err := entry.Validate(); err != nil {
 			return fmt.Errorf("playlist entry %d: %w", index, err)

internal/playback/playlist_controller.go

@@ -7,10 +7,22 @@ import (
 	"time"
 )
 
-type PlaylistReadiness struct {
+type PlaylistEventKind uint8
+
+const (
+	PlaylistEventReady PlaylistEventKind = iota
+	PlaylistEventFailed
+)
+
+type PlaylistEvent struct {
 	Revision uint64
+	Kind     PlaylistEventKind
 }
 
+// PlaylistReadiness is retained as an alias for callers that only publish
+// ready events. Its zero Kind is PlaylistEventReady.
+type PlaylistReadiness = PlaylistEvent
+
 type playlistTimer interface {
 	C() <-chan time.Time
 	Stop() bool
@@ -162,6 +174,40 @@ func (c *PlaylistController) Run(
 				readiness = nil
 				continue
 			}
+			if ready.Kind == PlaylistEventFailed {
+				if ready.Revision != revision {
+					continue
+				}
+				stopTimer()
+				// A failed entry must not retain a live or apparently active
+				// duration clock, even when the policy is to wait.
+				timing = NewPlaylistTiming(revision, timing.Duration)
+				if c.playlist.OnFailure != PlaylistFailureNext {
+					c.publish(state, revision, timing)
+					continue
+				}
+				next, sessionCommand, apply, err := ApplyPlaylistSelection(
+					c.playlist,
+					state,
+					PlaylistCommand{Kind: PlaylistNext},
+					c.retry,
+				)
+				if err != nil || !apply {
+					c.publish(state, revision, timing)
+					continue
+				}
+				select {
+				case <-ctx.Done():
+					return ctx.Err()
+				case c.sessions <- sessionCommand:
+				}
+				revision++
+				state = next
+				entry, _ := next.Entry(c.playlist)
+				timing = NewPlaylistTiming(revision, entry.Duration)
+				c.publish(state, revision, timing)
+				continue
+			}
 			if timing.Paused &&
 				ready.Revision == timing.Revision &&
 				timing.Duration > 0 &&

internal/playback/playlist_failure_test.go

@@ -0,0 +1,181 @@
+package playback
+
+import (
+	"context"
+	"testing"
+	"time"
+)
+
+func TestIsSessionFailed(t *testing.T) {
+	const generation = 7
+	video := FeedConfig{Domain: "/video", UUID: "video", Active: true}
+	audio := FeedConfig{Domain: "/audio", UUID: "audio", Active: true}
+	failedFeed := func(unit Unit, feed FeedConfig) Status {
+		return Status{
+			Unit: unit, State: StateFailed, Generation: generation, Feed: feed,
+		}
+	}
+	independent := SessionSnapshot{
+		Generation: generation,
+		Plan: SessionPlan{
+			Topology: TopologyIndependent,
+			Video:    video,
+			Audio:    audio,
+		},
+	}
+	synchronized := SessionSnapshot{
+		Generation: generation,
+		Plan: SessionPlan{
+			Topology: TopologySynchronized,
+			Sync:     SyncPairConfig{Video: video, Audio: audio},
+		},
+	}
+
+	tests := []struct {
+		name     string
+		session  SessionSnapshot
+		statuses PlaybackStatusSnapshot
+		want     bool
+	}{
+		{
+			name:    "video failure in independent pair",
+			session: independent,
+			statuses: PlaybackStatusSnapshot{
+				Generation: generation,
+				Video:      failedFeed(UnitVideo, video), HasVideo: true,
+			},
+			want: true,
+		},
+		{
+			name:    "audio failure in independent pair",
+			session: independent,
+			statuses: PlaybackStatusSnapshot{
+				Generation: generation,
+				Audio:      failedFeed(UnitAudio, audio), HasAudio: true,
+			},
+			want: true,
+		},
+		{
+			name:    "sync failure",
+			session: synchronized,
+			statuses: PlaybackStatusSnapshot{
+				Generation: generation,
+				Sync: Status{
+					Unit: UnitSync, State: StateFailed, Generation: generation,
+					Pair: SyncPairConfig{Video: video, Audio: audio},
+				},
+				HasSync: true,
+			},
+			want: true,
+		},
+		{
+			name:    "stale snapshot generation",
+			session: independent,
+			statuses: PlaybackStatusSnapshot{
+				Generation: generation - 1,
+				Video:      failedFeed(UnitVideo, video), HasVideo: true,
+			},
+		},
+		{
+			name:    "wrong source",
+			session: independent,
+			statuses: PlaybackStatusSnapshot{
+				Generation: generation,
+				Video: failedFeed(UnitVideo, FeedConfig{
+					Domain: "/video", UUID: "other", Active: true,
+				}),
+				HasVideo: true,
+			},
+		},
+		{
+			name: "inactive failed unit is ignored",
+			session: SessionSnapshot{
+				Generation: generation,
+				Plan: SessionPlan{
+					Topology: TopologyIndependent,
+					Video:    video,
+					Audio:    FeedConfig{Domain: audio.Domain, UUID: audio.UUID},
+				},
+			},
+			statuses: PlaybackStatusSnapshot{
+				Generation: generation,
+				Audio:      failedFeed(UnitAudio, audio), HasAudio: true,
+			},
+		},
+		{
+			name:    "reconnecting has not exhausted retries",
+			session: independent,
+			statuses: PlaybackStatusSnapshot{
+				Generation: generation,
+				Video: Status{
+					Unit: UnitVideo, State: StateReconnecting,
+					Generation: generation, Feed: video,
+				},
+				HasVideo: true,
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			if got := IsSessionFailed(test.session, test.statuses); got != test.want {
+				t.Fatalf("IsSessionFailed() = %v, want %v", got, test.want)
+			}
+		})
+	}
+}
+
+func TestPlaylistControllerFailurePolicy(t *testing.T) {
+	for _, test := range []struct {
+		name        string
+		policy      PlaylistFailurePolicy
+		wantAdvance bool
+	}{
+		{name: "wait", policy: PlaylistFailureWait},
+		{name: "next", policy: PlaylistFailureNext, wantAdvance: true},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			playlist := navigationPlaylist(false)
+			playlist.OnFailure = test.policy
+			sessions := make(chan SessionCommand, 4)
+			controller, err := NewPlaylistController(
+				playlist, validPlaylistRetryPolicy(), sessions,
+			)
+			if err != nil {
+				t.Fatalf("NewPlaylistController() error = %v", err)
+			}
+			commands := make(chan PlaylistCommand, 2)
+			events := make(chan PlaylistReadiness, 2)
+			ctx, cancel := context.WithCancel(context.Background())
+			result := make(chan error, 1)
+			go func() { result <- controller.Run(ctx, commands, events) }()
+
+			commands <- PlaylistCommand{Kind: PlaylistSelect, Index: 0}
+			<-sessions
+			events <- PlaylistEvent{Revision: 1, Kind: PlaylistEventFailed}
+
+			if test.wantAdvance {
+				select {
+				case <-sessions:
+				case <-time.After(time.Second):
+					t.Fatal("failure did not advance playlist")
+				}
+				snapshot, _ := controller.Snapshot()
+				if snapshot.State.CurrentIndex != 1 || snapshot.Revision != 2 {
+					t.Fatalf("snapshot = %#v, want index 1 revision 2", snapshot)
+				}
+			} else {
+				select {
+				case command := <-sessions:
+					t.Fatalf("unexpected session command: %#v", command)
+				case <-time.After(20 * time.Millisecond):
+				}
+			}
+
+			cancel()
+			if err := <-result; err != context.Canceled {
+				t.Fatalf("Run() error = %v, want context canceled", err)
+			}
+		})
+	}
+}

internal/playback/playlist_readiness.go

@@ -89,7 +89,8 @@ func (c *PlaylistReadinessCoordinator) Run(ctx context.Context) error {
 	ticker := c.newTicker(c.interval)
 	defer ticker.Stop()
 
-	var emittedRevision uint64
+	var emittedReadyRevision uint64
+	var emittedFailedRevision uint64
 	for {
 		select {
 		case <-ctx.Done():
@@ -99,11 +100,7 @@ func (c *PlaylistReadinessCoordinator) Run(ctx context.Context) error {
 			playlistSnapshot, ok := c.playlist.Snapshot()
 			if !ok ||
 				!playlistSnapshot.State.HasSelection ||
-				playlistSnapshot.Revision == 0 ||
+				playlistSnapshot.Revision == 0 {
-				playlistSnapshot.Entry.Duration <= 0 ||
-				playlistSnapshot.Timing.Started ||
-				playlistSnapshot.Timing.Paused ||
-				playlistSnapshot.Revision == emittedRevision {
 				continue
 			}
 
@@ -114,7 +111,30 @@ func (c *PlaylistReadinessCoordinator) Run(ctx context.Context) error {
 			) {
 				continue
 			}
-			if !IsSessionPlaying(sessionSnapshot, c.statuses.SnapshotAll()) {
+			statuses := c.statuses.SnapshotAll()
+			if IsSessionFailed(sessionSnapshot, statuses) {
+				if playlistSnapshot.Revision == emittedFailedRevision {
+					continue
+				}
+				failed := PlaylistEvent{
+					Revision: playlistSnapshot.Revision,
+					Kind:     PlaylistEventFailed,
+				}
+				select {
+				case <-ctx.Done():
+					return ctx.Err()
+				case c.output <- failed:
+					emittedFailedRevision = playlistSnapshot.Revision
+				}
+				continue
+			}
+			if playlistSnapshot.Entry.Duration <= 0 ||
+				playlistSnapshot.Timing.Started ||
+				playlistSnapshot.Timing.Paused ||
+				playlistSnapshot.Revision == emittedReadyRevision {
+				continue
+			}
+			if !IsSessionPlaying(sessionSnapshot, statuses) {
 				continue
 			}
 
@@ -123,7 +143,7 @@ func (c *PlaylistReadinessCoordinator) Run(ctx context.Context) error {
 			case <-ctx.Done():
 				return ctx.Err()
 			case c.output <- ready:
-				emittedRevision = playlistSnapshot.Revision
+				emittedReadyRevision = playlistSnapshot.Revision
 			}
 		}
 	}
@@ -193,6 +213,38 @@ func IsSessionPlaying(
 	}
 }
 
+func IsSessionFailed(
+	session SessionSnapshot,
+	statuses PlaybackStatusSnapshot,
+) bool {
+	if statuses.Generation != session.Generation {
+		return false
+	}
+
+	switch session.Plan.Topology {
+	case TopologyIndependent:
+		return (session.Plan.Video.Active && statusIsFailed(
+			statuses.Video, statuses.HasVideo, session.Generation, session.Plan.Video,
+		)) || (session.Plan.Audio.Active && statusIsFailed(
+			statuses.Audio, statuses.HasAudio, session.Generation, session.Plan.Audio,
+		))
+	case TopologySynchronized:
+		return statuses.HasSync &&
+			statuses.Sync.Generation == session.Generation &&
+			statuses.Sync.State == StateFailed &&
+			sameSyncSource(statuses.Sync.Pair, session.Plan.Sync)
+	default:
+		return false
+	}
+}
+
+func statusIsFailed(status Status, present bool, generation uint64, feed FeedConfig) bool {
+	return present &&
+		status.Generation == generation &&
+		status.State == StateFailed &&
+		sameFeedSource(status.Feed, feed)
+}
+
 func statusIsPlaying(
 	status Status,
 	present bool,

internal/playback/sync_attempt.go

@@ -14,6 +14,7 @@ func runSyncAttempt(
 	videoConfig FeedConfig,
 	audioConfig FeedConfig,
 ) (resultErr error) {
+	videoCtx := withVideoSource(ctx, videoConfig)
 	reader, err := factory.OpenSync(
 		ctx,
 		videoConfig,
@@ -39,7 +40,7 @@ func runSyncAttempt(
 			return fmt.Errorf("read sync group: %w", err)
 		}
 
-		if err := videoSink.ConsumeVideo(ctx, frame.Video); err != nil {
+		if err := videoSink.ConsumeVideo(videoCtx, frame.Video); err != nil {
 			if ctx.Err() != nil {
 				return ctx.Err()
 			}

internal/playback/video.go

@@ -2,6 +2,17 @@ package playback
 
 import "context"
 
+type videoSourceContextKey struct{}
+
+func withVideoSource(ctx context.Context, source FeedConfig) context.Context {
+	return context.WithValue(ctx, videoSourceContextKey{}, source)
+}
+
+func videoSourceFromContext(ctx context.Context) FeedConfig {
+	source, _ := ctx.Value(videoSourceContextKey{}).(FeedConfig)
+	return source
+}
+
 // VideoFrame contains metadata and borrowed source payload.
 //
 // Payload is valid only until the next VideoReader.ReadVideo call or until the

internal/playback/video_attempt.go

@@ -24,6 +24,7 @@ func runVideoAttempt(
 	sink VideoSink,
 	config FeedConfig,
 ) (resultErr error) {
+	ctx = withVideoSource(ctx, config)
 	reader, err := factory.OpenVideo(ctx, config)
 	if err != nil {
 		return fmt.Errorf("open video: %w", err)

internal/playback/video_bridge.go

@@ -6,7 +6,9 @@ import (
 )
 
 type PendingVideoFrame struct {
-	Frame VideoFrame
+	Frame      VideoFrame
+	Generation uint64
+	Source     FeedConfig
 
 	completeOnce sync.Once
 	result       chan error
@@ -27,8 +29,10 @@ func (b *VideoBridge) ConsumeVideo(
 	frame VideoFrame,
 ) error {
 	pending := &PendingVideoFrame{
-		Frame:  frame,
+		Frame:      frame,
-		result: make(chan error, 1),
+		Generation: generationFromContext(ctx),
+		Source:     videoSourceFromContext(ctx),
+		result:     make(chan error, 1),
 	}
 
 	select {

internal/playback/video_bridge_test.go

@@ -21,8 +21,10 @@ func TestVideoBridgeDeliversFrameAndCompletionResult(t *testing.T) {
 	}
 	consumeResult := make(chan error, 1)
 
+	wantSource := FeedConfig{Domain: "/video", UUID: "video", Active: true}
 	go func() {
-		consumeResult <- bridge.ConsumeVideo(context.Background(), wantFrame)
+		ctx := withGeneration(context.Background(), 17)
+		consumeResult <- bridge.ConsumeVideo(withVideoSource(ctx, wantSource), wantFrame)
 	}()
 
 	ctx, cancel := context.WithTimeout(context.Background(), videoBridgeTestTimeout)
@@ -34,6 +36,12 @@ func TestVideoBridgeDeliversFrameAndCompletionResult(t *testing.T) {
 	if pending.Frame.Index != wantFrame.Index {
 		t.Fatalf("Next() frame index = %d, want %d", pending.Frame.Index, wantFrame.Index)
 	}
+	if pending.Generation != 17 {
+		t.Fatalf("Next() generation = %d, want 17", pending.Generation)
+	}
+	if pending.Source != wantSource {
+		t.Fatalf("Next() source = %#v, want %#v", pending.Source, wantSource)
+	}
 	if &pending.Frame.Payload[0] != &wantFrame.Payload[0] {
 		t.Fatal("Next() copied the borrowed payload")
 	}

playlists/sample-list-all.json

@@ -1,5 +1,6 @@
 {
   "loop": true,
+  "on_failure": "next",
   "entries": [
     {
       "name": "timelapse",

playlists/sample-list.json

@@ -1,5 +1,6 @@
 {
   "loop": true,
+  "on_failure": "next",
   "entries": [
     {
       "name": "timelapse",
@@ -14,6 +15,19 @@
       "sync": true,
       "duration": "10s"
     },
+    {
+      "name": "fail",
+      "video": {
+        "domain": "/dev/shm/mxl",
+        "uuid": "6fbec3b1-1b0f-417d-9059-8b94a47197ed"
+      },
+      "audio": {
+        "domain": "/dev/shm/mxl",
+        "uuid": "6fbec3b1-1b0f-417d-9059-8b94a47197ec"
+      },
+      "sync": true,
+      "duration": "10s"
+    },
     {
       "name": "F1 Highlights",
       "video": {