feeds-uuid.txt

@@ -0,0 +1,8 @@
+5fbec3b1-1b0f-417d-9059-8b94a47197ed
+5fbec3b1-1b0f-417d-9059-8b94a47197ec
+
+5fbec3b1-1b0f-417d-9059-8b94a47197ef
+5fbec3b1-1b0f-417d-9059-8b94a47197eb
+
+2618979d-76a5-45e0-83cb-0f192978d1cd
+9d2a041b-01cf-4ee4-bffa-188fe093c99b

internal/playback/playlist_pipeline_test.go

@@ -0,0 +1,175 @@
+package playback
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+)
+
+func TestPlaylistFailurePipeline(t *testing.T) {
+	for _, test := range []struct {
+		name        string
+		policy      PlaylistFailurePolicy
+		wantAdvance bool
+	}{
+		{name: "wait retains failed entry", policy: PlaylistFailureWait},
+		{name: "next advances failed entry", policy: PlaylistFailureNext, wantAdvance: true},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			playlist := Playlist{
+				OnFailure: test.policy,
+				Entries: []PlaylistEntry{
+					{
+						Name:  "independent",
+						Video: PlaylistFeed{Domain: "/video", UUID: "video"},
+						Audio: PlaylistFeed{Domain: "/audio", UUID: "audio"},
+					},
+					{Name: "next", Video: PlaylistFeed{Domain: "/video", UUID: "next"}},
+				},
+			}
+			h := startPlaylistPipeline(t, playlist)
+			defer h.stop(t)
+
+			h.commands <- PlaylistCommand{Kind: PlaylistSelect, Index: 0}
+			selected := receivePlaylistSession(t, h.sessions).Session
+			waitForPlaylistSnapshot(t, h.controller, func(snapshot PlaylistSnapshot) bool {
+				return snapshot.Revision == 1
+			})
+			h.setSession(selected, 4)
+			failureErr := errors.New("audio retries exhausted")
+			h.statuses.set(PlaybackStatusSnapshot{
+				Generation: 4,
+				Audio: Status{
+					Unit: UnitAudio, State: StateFailed, Generation: 4,
+					Feed: selected.Audio, Attempt: 3, FailedAttempts: 3, Err: failureErr,
+				},
+				HasAudio: true,
+			})
+			h.ticker.tick()
+
+			if test.wantAdvance {
+				next := receivePlaylistSession(t, h.sessions)
+				if next.Session.Video.UUID != "next" || next.Session.Audio.IsConfigured() {
+					t.Fatalf("advanced session = %#v", next.Session)
+				}
+			} else {
+				select {
+				case command := <-h.sessions:
+					t.Fatalf("wait policy advanced with %#v", command)
+				case <-time.After(20 * time.Millisecond):
+				}
+			}
+
+			snapshot := waitForPlaylistSnapshot(t, h.controller, func(snapshot PlaylistSnapshot) bool {
+				return snapshot.HasFailure
+			})
+			if snapshot.Failure.Status.Unit != UnitAudio ||
+				!errors.Is(snapshot.Failure.Status.Err, failureErr) {
+				t.Fatalf("failure snapshot = %#v", snapshot.Failure)
+			}
+			wantIndex := 0
+			if test.wantAdvance {
+				wantIndex = 1
+			}
+			if snapshot.State.CurrentIndex != wantIndex {
+				t.Fatalf("current index = %d, want %d", snapshot.State.CurrentIndex, wantIndex)
+			}
+		})
+	}
+}
+
+func TestPlaylistManualSelectionWhileFeedIsReconnecting(t *testing.T) {
+	playlist := navigationPlaylist(false)
+	h := startPlaylistPipeline(t, playlist)
+	defer h.stop(t)
+
+	h.commands <- PlaylistCommand{Kind: PlaylistSelect, Index: 0}
+	selected := receivePlaylistSession(t, h.sessions).Session
+	waitForPlaylistSnapshot(t, h.controller, func(snapshot PlaylistSnapshot) bool {
+		return snapshot.Revision == 1
+	})
+	h.setSession(selected, 2)
+	h.statuses.set(PlaybackStatusSnapshot{
+		Generation: 2,
+		Video: Status{
+			Unit: UnitVideo, State: StateReconnecting, Generation: 2,
+			Feed: selected.Video, Attempt: 2, FailedAttempts: 1,
+		},
+		HasVideo: true,
+	})
+	h.ticker.tick()
+
+	h.commands <- PlaylistCommand{Kind: PlaylistSelect, Index: 1}
+	next := receivePlaylistSession(t, h.sessions)
+	if next.Session.Audio.UUID != "audio-2" || next.Session.Video.IsConfigured() {
+		t.Fatalf("manual selection session = %#v", next.Session)
+	}
+	waitForPlaylistSnapshot(t, h.controller, func(snapshot PlaylistSnapshot) bool {
+		return snapshot.Revision == 2 && snapshot.State.CurrentIndex == 1
+	})
+}
+
+type playlistPipelineHarness struct {
+	controller *PlaylistController
+	commands   chan PlaylistCommand
+	sessions   chan SessionCommand
+	session    *fakeSessionSnapshotSource
+	statuses   *fakePlaybackStatusSnapshotSource
+	ticker     *fakePlaylistReadinessTicker
+	cancel     context.CancelFunc
+	results    chan error
+}
+
+func startPlaylistPipeline(t *testing.T, playlist Playlist) *playlistPipelineHarness {
+	t.Helper()
+	sessions := make(chan SessionCommand, 8)
+	controller, err := NewPlaylistController(playlist, validPlaylistRetryPolicy(), sessions)
+	if err != nil {
+		t.Fatal(err)
+	}
+	events := make(chan PlaylistReadiness, 8)
+	session := &fakeSessionSnapshotSource{}
+	statuses := &fakePlaybackStatusSnapshotSource{}
+	coordinator, err := NewPlaylistReadinessCoordinator(
+		controller, session, statuses, events, time.Millisecond,
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	ticker := newFakePlaylistReadinessTicker()
+	coordinator.newTicker = func(time.Duration) playlistReadinessTicker { return ticker }
+	commands := make(chan PlaylistCommand, 8)
+	ctx, cancel := context.WithCancel(context.Background())
+	results := make(chan error, 2)
+	go func() { results <- controller.Run(ctx, commands, events) }()
+	go func() { results <- coordinator.Run(ctx) }()
+	return &playlistPipelineHarness{
+		controller: controller, commands: commands, sessions: sessions,
+		session: session, statuses: statuses, ticker: ticker,
+		cancel: cancel, results: results,
+	}
+}
+
+func (h *playlistPipelineHarness) setSession(desired SessionConfig, generation uint64) {
+	plan, err := BuildSessionPlan(desired, func(FeedConfig, FeedConfig) bool { return false })
+	if err != nil {
+		panic(err)
+	}
+	h.session.set(SessionSnapshot{Desired: desired, Plan: plan, Generation: generation}, true)
+}
+
+func (h *playlistPipelineHarness) stop(t *testing.T) {
+	t.Helper()
+	h.cancel()
+	for range 2 {
+		select {
+		case err := <-h.results:
+			if !errors.Is(err, context.Canceled) {
+				t.Fatalf("pipeline Run() error = %v", err)
+			}
+		case <-time.After(time.Second):
+			t.Fatal("playlist pipeline did not stop")
+		}
+	}
+}

internal/playback/session_controller_test.go

@@ -344,6 +344,38 @@ func TestSessionControllerCancellationStopsAndJoinsRuntime(t *testing.T) {
 	}
 }
 
+func TestSessionControllerStopAllWhileSessionIsActive(t *testing.T) {
+	events := make(chan controllerEvent, 32)
+	controller := newRecordingController(t, events)
+	initial := validCommandSession()
+	initial.SyncRequested = false
+	commands := make(chan SessionCommand)
+	done := make(chan error, 1)
+	go func() { done <- controller.Run(context.Background(), initial, commands) }()
+	receiveIndependentStarts(t, events)
+
+	commands <- SessionCommand{Kind: CommandStopAll}
+	stopped := map[Unit]bool{}
+	for len(stopped) < 2 {
+		event := receiveControllerEvent(t, events)
+		if event.action != "stop" || (event.unit != UnitVideo && event.unit != UnitAudio) {
+			t.Fatalf("unexpected stop event: %+v", event)
+		}
+		stopped[event.unit] = true
+	}
+	snapshot := waitControllerSnapshot(t, controller, func(snapshot SessionSnapshot) bool {
+		return snapshot.Plan.Topology == TopologyIdle
+	})
+	if snapshot.Desired.Video.Active || snapshot.Desired.Audio.Active {
+		t.Fatalf("stopped desired session = %#v", snapshot.Desired)
+	}
+
+	close(commands)
+	if err := <-done; err != nil {
+		t.Fatalf("Run() error = %v", err)
+	}
+}
+
 func TestSessionControllerSnapshotTracksDesiredPlanAndGeneration(t *testing.T) {
 	events := make(chan controllerEvent, 64)
 	controller := newRecordingController(t, events)