diff --git a/internal/playback/playlist_controller.go b/internal/playback/playlist_controller.go index d9f9c87..3cf6825 100644 --- a/internal/playback/playlist_controller.go +++ b/internal/playback/playlist_controller.go @@ -4,12 +4,33 @@ import ( "context" "errors" "sync" + "time" ) +type PlaylistReadiness struct { + Revision uint64 +} + +type playlistTimer interface { + C() <-chan time.Time + Stop() bool +} + +type playlistTimerFactory func(time.Duration) playlistTimer + +type realPlaylistTimer struct { + timer *time.Timer +} + +func (t realPlaylistTimer) C() <-chan time.Time { return t.timer.C } +func (t realPlaylistTimer) Stop() bool { return t.timer.Stop() } + type PlaylistController struct { playlist Playlist retry RetryPolicy sessions chan<- SessionCommand + now func() time.Time + newTimer playlistTimerFactory mu sync.RWMutex snapshot PlaylistSnapshot @@ -20,6 +41,7 @@ type PlaylistSnapshot struct { State PlaylistState Entry PlaylistEntry Revision uint64 + Timing PlaylistTimingState } var ( @@ -44,16 +66,32 @@ func NewPlaylistController( playlist: playlist, retry: retry, sessions: sessions, + now: time.Now, + newTimer: func(duration time.Duration) playlistTimer { + return realPlaylistTimer{timer: time.NewTimer(duration)} + }, }, nil } func (c *PlaylistController) Run( ctx context.Context, commands <-chan PlaylistCommand, + readiness <-chan PlaylistReadiness, ) error { state := PlaylistState{} revision := uint64(0) - c.publish(state, revision) + timing := PlaylistTimingState{} + var timer playlistTimer + var timerC <-chan time.Time + var timerRevision uint64 + c.publish(state, revision, timing) + + stopTimer := func() { + stopPlaylistTimer(timer) + timer = nil + timerC = nil + } + defer stopTimer() for { select { @@ -75,16 +113,75 @@ func (c *PlaylistController) Run( continue } if apply { + stopTimer() select { case <-ctx.Done(): return ctx.Err() case c.sessions <- sessionCommand: } revision++ + entry, _ := next.Entry(c.playlist) + timing = NewPlaylistTiming(revision, entry.Duration) } state = next - c.publish(state, revision) + c.publish(state, revision, timing) + + case ready, ok := <-readiness: + if !ok { + readiness = nil + continue + } + nextTiming, started := StartPlaylistTiming( + timing, + ready.Revision, + c.now(), + ) + if !started { + continue + } + timing = nextTiming + timerRevision = timing.Revision + timer = c.newTimer(timing.Duration) + timerC = timer.C() + c.publish(state, revision, timing) + + case firedAt := <-timerC: + firedRevision := timerRevision + timer = nil + timerC = nil + nextTiming, expired := ExpirePlaylistTiming( + timing, + firedRevision, + firedAt, + ) + if !expired { + continue + } + timing = nextTiming + + next, sessionCommand, apply, err := ApplyPlaylistSelection( + c.playlist, + state, + PlaylistCommand{Kind: PlaylistNext}, + c.retry, + ) + if err != nil { + c.publish(state, revision, timing) + continue + } + if apply { + select { + case <-ctx.Done(): + return ctx.Err() + case c.sessions <- sessionCommand: + } + revision++ + entry, _ := next.Entry(c.playlist) + timing = NewPlaylistTiming(revision, entry.Duration) + } + state = next + c.publish(state, revision, timing) } } } @@ -95,7 +192,11 @@ func (c *PlaylistController) Snapshot() (PlaylistSnapshot, bool) { return c.snapshot, c.hasSnapshot } -func (c *PlaylistController) publish(state PlaylistState, revision uint64) { +func (c *PlaylistController) publish( + state PlaylistState, + revision uint64, + timing PlaylistTimingState, +) { entry, _ := state.Entry(c.playlist) c.mu.Lock() @@ -103,7 +204,18 @@ func (c *PlaylistController) publish(state PlaylistState, revision uint64) { State: state, Entry: entry, Revision: revision, + Timing: timing, } c.hasSnapshot = true c.mu.Unlock() } + +func stopPlaylistTimer(timer playlistTimer) { + if timer == nil || timer.Stop() { + return + } + select { + case <-timer.C(): + default: + } +} diff --git a/internal/playback/playlist_controller_test.go b/internal/playback/playlist_controller_test.go index 31ffddb..886b055 100644 --- a/internal/playback/playlist_controller_test.go +++ b/internal/playback/playlist_controller_test.go @@ -173,7 +173,7 @@ func TestPlaylistControllerCommitsStateAfterSessionDelivery(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() result := make(chan error, 1) - go func() { result <- controller.Run(ctx, commands) }() + go func() { result <- controller.Run(ctx, commands, nil) }() waitForPlaylistSnapshot(t, controller, func(snapshot PlaylistSnapshot) bool { return !snapshot.State.HasSelection @@ -224,7 +224,7 @@ func TestPlaylistControllerCancellationWhileSending(t *testing.T) { commands := make(chan PlaylistCommand, 1) ctx, cancel := context.WithCancel(context.Background()) result := make(chan error, 1) - go func() { result <- controller.Run(ctx, commands) }() + go func() { result <- controller.Run(ctx, commands, nil) }() commands <- PlaylistCommand{Kind: PlaylistNext} cancel() @@ -323,7 +323,7 @@ func startPlaylistController( commands := make(chan PlaylistCommand, 64) ctx, cancel := context.WithCancel(context.Background()) result := make(chan error, 1) - go func() { result <- controller.Run(ctx, commands) }() + go func() { result <- controller.Run(ctx, commands, nil) }() return controller, commands, sessions, cancel, result } diff --git a/internal/playback/playlist_controller_timing_test.go b/internal/playback/playlist_controller_timing_test.go new file mode 100644 index 0000000..3cdc502 --- /dev/null +++ b/internal/playback/playlist_controller_timing_test.go @@ -0,0 +1,311 @@ +package playback + +import ( + "context" + "errors" + "sync" + "testing" + "time" +) + +type fakePlaylistTimer struct { + ch chan time.Time + + mu sync.Mutex + stopped bool +} + +func newFakePlaylistTimer() *fakePlaylistTimer { + return &fakePlaylistTimer{ch: make(chan time.Time, 1)} +} + +func (t *fakePlaylistTimer) C() <-chan time.Time { return t.ch } + +func (t *fakePlaylistTimer) Stop() bool { + t.mu.Lock() + defer t.mu.Unlock() + alreadyStopped := t.stopped + t.stopped = true + return !alreadyStopped +} + +func (t *fakePlaylistTimer) isStopped() bool { + t.mu.Lock() + defer t.mu.Unlock() + return t.stopped +} + +func (t *fakePlaylistTimer) fire(at time.Time) { + t.ch <- at +} + +func timedPlaylist(loop bool) Playlist { + return Playlist{ + Entries: []PlaylistEntry{ + { + Name: "first", + Video: PlaylistFeed{Domain: "domain", UUID: "video-1"}, + Duration: 10 * time.Second, + }, + { + Name: "second", + Audio: PlaylistFeed{Domain: "domain", UUID: "audio-2"}, + Duration: 20 * time.Second, + }, + }, + Loop: loop, + } +} + +func TestPlaylistControllerStartsOneTimerForMatchingReadiness(t *testing.T) { + controller, commands, readiness, sessions, timers, now, cancel, result := + startTimedPlaylistController(t, timedPlaylist(false)) + defer cancel() + + commands <- PlaylistCommand{Kind: PlaylistNext} + _ = receivePlaylistSession(t, sessions) + waitForPlaylistSnapshot(t, controller, func(snapshot PlaylistSnapshot) bool { + return snapshot.Revision == 1 + }) + + readiness <- PlaylistReadiness{Revision: 0} + assertNoPlaylistTimer(t, timers) + readiness <- PlaylistReadiness{Revision: 1} + timer := receiveFakePlaylistTimer(t, timers) + + snapshot := waitForPlaylistSnapshot(t, controller, func(snapshot PlaylistSnapshot) bool { + return snapshot.Timing.Started + }) + if snapshot.Timing.Deadline != now.Add(10*time.Second) { + t.Fatalf("deadline = %v, want %v", snapshot.Timing.Deadline, now.Add(10*time.Second)) + } + readiness <- PlaylistReadiness{Revision: 1} + assertNoPlaylistTimer(t, timers) + if timer.isStopped() { + t.Fatal("timer stopped after duplicate readiness") + } + + close(commands) + if err := waitForPlaylistResult(t, result); err != nil { + t.Fatalf("Run() error = %v", err) + } + if !timer.isStopped() { + t.Fatal("timer was not stopped when commands closed") + } +} + +func TestPlaylistControllerTimerExpiryAdvances(t *testing.T) { + controller, commands, readiness, sessions, timers, now, cancel, result := + startTimedPlaylistController(t, timedPlaylist(false)) + defer cancel() + + commands <- PlaylistCommand{Kind: PlaylistNext} + _ = receivePlaylistSession(t, sessions) + readiness <- PlaylistReadiness{Revision: 1} + timer := receiveFakePlaylistTimer(t, timers) + timer.fire(now.Add(10 * time.Second)) + + session := receivePlaylistSession(t, sessions) + if session.Session.Audio.UUID != "audio-2" || session.Session.Video.IsConfigured() { + t.Fatalf("advanced session = %#v, want audio-only second entry", session.Session) + } + snapshot := waitForPlaylistSnapshot(t, controller, func(snapshot PlaylistSnapshot) bool { + return snapshot.Revision == 2 + }) + if snapshot.State.CurrentIndex != 1 || snapshot.Timing.Started { + t.Fatalf("advanced snapshot = %#v", snapshot) + } + if snapshot.Timing.Duration != 20*time.Second { + t.Fatalf("next duration = %v, want %v", snapshot.Timing.Duration, 20*time.Second) + } + + close(commands) + if err := waitForPlaylistResult(t, result); err != nil { + t.Fatalf("Run() error = %v", err) + } +} + +func TestPlaylistControllerLoopingTimerExpiryWraps(t *testing.T) { + controller, commands, readiness, sessions, timers, now, cancel, result := + startTimedPlaylistController(t, timedPlaylist(true)) + defer cancel() + + commands <- PlaylistCommand{Kind: PlaylistSelect, Index: 1} + _ = receivePlaylistSession(t, sessions) + readiness <- PlaylistReadiness{Revision: 1} + timer := receiveFakePlaylistTimer(t, timers) + waitForPlaylistSnapshot(t, controller, func(snapshot PlaylistSnapshot) bool { + return snapshot.Timing.Started + }) + timer.fire(now.Add(20 * time.Second)) + + session := receivePlaylistSession(t, sessions) + if session.Session.Video.UUID != "video-1" { + t.Fatalf("wrapped session = %#v, want first entry", session.Session) + } + waitForPlaylistSnapshot(t, controller, func(snapshot PlaylistSnapshot) bool { + return snapshot.Revision == 2 && snapshot.State.CurrentIndex == 0 + }) + + close(commands) + if err := waitForPlaylistResult(t, result); err != nil { + t.Fatalf("Run() error = %v", err) + } +} + +func TestPlaylistControllerFinalExpiryStopsWithoutCommand(t *testing.T) { + controller, commands, readiness, sessions, timers, now, cancel, result := + startTimedPlaylistController(t, timedPlaylist(false)) + defer cancel() + + commands <- PlaylistCommand{Kind: PlaylistSelect, Index: 1} + _ = receivePlaylistSession(t, sessions) + readiness <- PlaylistReadiness{Revision: 1} + timer := receiveFakePlaylistTimer(t, timers) + waitForPlaylistSnapshot(t, controller, func(snapshot PlaylistSnapshot) bool { + return snapshot.Timing.Started + }) + timer.fire(now.Add(20 * time.Second)) + + snapshot := waitForPlaylistSnapshot(t, controller, func(snapshot PlaylistSnapshot) bool { + return snapshot.Revision == 1 && !snapshot.Timing.Started + }) + if snapshot.State.CurrentIndex != 1 { + t.Fatalf("final snapshot state = %#v, want final entry", snapshot.State) + } + select { + case command := <-sessions: + t.Fatalf("unexpected session command: %#v", command) + case <-time.After(20 * time.Millisecond): + } + + close(commands) + if err := waitForPlaylistResult(t, result); err != nil { + t.Fatalf("Run() error = %v", err) + } +} + +func TestPlaylistControllerManualSelectionStopsOldTimer(t *testing.T) { + controller, commands, readiness, sessions, timers, now, cancel, result := + startTimedPlaylistController(t, timedPlaylist(false)) + defer cancel() + + commands <- PlaylistCommand{Kind: PlaylistNext} + _ = receivePlaylistSession(t, sessions) + readiness <- PlaylistReadiness{Revision: 1} + oldTimer := receiveFakePlaylistTimer(t, timers) + + commands <- PlaylistCommand{Kind: PlaylistSelect, Index: 1} + _ = receivePlaylistSession(t, sessions) + waitForPlaylistSnapshot(t, controller, func(snapshot PlaylistSnapshot) bool { + return snapshot.Revision == 2 + }) + if !oldTimer.isStopped() { + t.Fatal("old timer was not stopped by manual selection") + } + oldTimer.fire(now.Add(10 * time.Second)) + select { + case command := <-sessions: + t.Fatalf("stale timer produced session command: %#v", command) + case <-time.After(20 * time.Millisecond): + } + snapshot, _ := controller.Snapshot() + if snapshot.Revision != 2 || snapshot.State.CurrentIndex != 1 { + t.Fatalf("stale timer changed snapshot: %#v", snapshot) + } + + close(commands) + if err := waitForPlaylistResult(t, result); err != nil { + t.Fatalf("Run() error = %v", err) + } +} + +func TestPlaylistControllerZeroDurationDoesNotCreateTimer(t *testing.T) { + playlist := timedPlaylist(false) + playlist.Entries[0].Duration = 0 + _, commands, readiness, sessions, timers, _, cancel, result := + startTimedPlaylistController(t, playlist) + defer cancel() + + commands <- PlaylistCommand{Kind: PlaylistNext} + _ = receivePlaylistSession(t, sessions) + readiness <- PlaylistReadiness{Revision: 1} + assertNoPlaylistTimer(t, timers) + + close(commands) + if err := waitForPlaylistResult(t, result); err != nil { + t.Fatalf("Run() error = %v", err) + } +} + +func TestPlaylistControllerCancellationStopsTimer(t *testing.T) { + _, commands, readiness, sessions, timers, _, cancel, result := + startTimedPlaylistController(t, timedPlaylist(false)) + + commands <- PlaylistCommand{Kind: PlaylistNext} + _ = receivePlaylistSession(t, sessions) + readiness <- PlaylistReadiness{Revision: 1} + timer := receiveFakePlaylistTimer(t, timers) + cancel() + if err := waitForPlaylistResult(t, result); !errors.Is(err, context.Canceled) { + t.Fatalf("Run() error = %v, want %v", err, context.Canceled) + } + if !timer.isStopped() { + t.Fatal("timer was not stopped on cancellation") + } +} + +func startTimedPlaylistController( + t *testing.T, + playlist Playlist, +) ( + *PlaylistController, + chan PlaylistCommand, + chan PlaylistReadiness, + chan SessionCommand, + chan *fakePlaylistTimer, + time.Time, + context.CancelFunc, + <-chan error, +) { + t.Helper() + sessions := make(chan SessionCommand, 16) + controller, err := NewPlaylistController(playlist, validPlaylistRetryPolicy(), sessions) + if err != nil { + t.Fatalf("NewPlaylistController() error = %v", err) + } + now := time.Date(2026, time.September, 1, 12, 0, 0, 0, time.UTC) + controller.now = func() time.Time { return now } + timers := make(chan *fakePlaylistTimer, 16) + controller.newTimer = func(time.Duration) playlistTimer { + timer := newFakePlaylistTimer() + timers <- timer + return timer + } + commands := make(chan PlaylistCommand, 16) + readiness := make(chan PlaylistReadiness, 16) + ctx, cancel := context.WithCancel(context.Background()) + result := make(chan error, 1) + go func() { result <- controller.Run(ctx, commands, readiness) }() + return controller, commands, readiness, sessions, timers, now, cancel, result +} + +func receiveFakePlaylistTimer(t *testing.T, timers <-chan *fakePlaylistTimer) *fakePlaylistTimer { + t.Helper() + select { + case timer := <-timers: + return timer + case <-time.After(time.Second): + t.Fatal("timed out waiting for playlist timer") + return nil + } +} + +func assertNoPlaylistTimer(t *testing.T, timers <-chan *fakePlaylistTimer) { + t.Helper() + select { + case timer := <-timers: + t.Fatalf("unexpected playlist timer: %#v", timer) + case <-time.After(20 * time.Millisecond): + } +}