From cd0298b136156bfb82574c250557643f09164009 Mon Sep 17 00:00:00 2001 From: Dmitry Sergeev Date: Tue, 1 Sep 2026 21:22:30 +0300 Subject: [PATCH] source-aware playback statuses. --- cmd/mxl-player/main.go | 2 +- internal/playback/audio_worker.go | 17 ++--- internal/playback/audio_worker_test.go | 9 +-- internal/playback/playlist_readiness.go | 30 ++++++-- internal/playback/playlist_readiness_test.go | 75 +++++++++++++++++++- internal/playback/state.go | 10 ++- internal/playback/sync_worker.go | 22 +++--- internal/playback/sync_worker_test.go | 4 ++ internal/playback/video_worker.go | 17 ++--- internal/playback/video_worker_test.go | 6 +- 10 files changed, 151 insertions(+), 41 deletions(-) diff --git a/cmd/mxl-player/main.go b/cmd/mxl-player/main.go index f6bdd06..8f78947 100644 --- a/cmd/mxl-player/main.go +++ b/cmd/mxl-player/main.go @@ -730,7 +730,7 @@ func main() { } } - if cimgui.CollapsingHeaderTreeNodeFlagsV("Debug Info", collapsingHeaderFlags) { + if cimgui.CollapsingHeaderTreeNodeFlagsV("Debug Info", cimgui.TreeNodeFlagsNone) { drawDebugSection() } } diff --git a/internal/playback/audio_worker.go b/internal/playback/audio_worker.go index e0ab115..bc8bc80 100644 --- a/internal/playback/audio_worker.go +++ b/internal/playback/audio_worker.go @@ -72,8 +72,9 @@ func (s *stabilityAudioSink) ConsumeAudio( return err } -func (w *AudioWorker) emit(ctx context.Context, status Status) { +func (w *AudioWorker) emit(ctx context.Context, config FeedConfig, status Status) { status.Generation = generationFromContext(ctx) + status.Feed = config if w.observer != nil { w.observer(status) } @@ -100,7 +101,7 @@ func (w *AudioWorker) Run( if attemptNumber > 1 { state = StateReconnecting } - w.emit(ctx, Status{ + w.emit(ctx, config, Status{ Unit: UnitAudio, State: state, Attempt: attemptNumber, @@ -109,7 +110,7 @@ func (w *AudioWorker) Run( attemptSink := &stabilityAudioSink{ sink: w.sink, onStable: func() { - w.emit(ctx, Status{ + w.emit(ctx, config, Status{ Unit: UnitAudio, State: StatePlaying, Attempt: attemptNumber, @@ -135,7 +136,7 @@ func (w *AudioWorker) Run( return } - w.emit(ctx, Status{ + w.emit(ctx, config, Status{ Unit: UnitAudio, State: StateReconnecting, Attempt: attemptNumber + 1, @@ -155,11 +156,11 @@ func (w *AudioWorker) Run( ) if ctx.Err() != nil { - w.emit(ctx, Status{ + w.emit(ctx, config, Status{ Unit: UnitAudio, State: StateStopping, }) - w.emit(ctx, Status{ + w.emit(ctx, config, Status{ Unit: UnitAudio, State: StateIdle, }) @@ -167,7 +168,7 @@ func (w *AudioWorker) Run( } if err != nil { - w.emit(ctx, Status{ + w.emit(ctx, config, Status{ Unit: UnitAudio, State: StateFailed, Attempt: attemptNumber, @@ -177,7 +178,7 @@ func (w *AudioWorker) Run( return err } - w.emit(ctx, Status{ + w.emit(ctx, config, Status{ Unit: UnitAudio, State: StateIdle, }) diff --git a/internal/playback/audio_worker_test.go b/internal/playback/audio_worker_test.go index a560ba1..7c78975 100644 --- a/internal/playback/audio_worker_test.go +++ b/internal/playback/audio_worker_test.go @@ -127,10 +127,8 @@ func TestAudioWorkerStatusesInheritGeneration(t *testing.T) { func(error) bool { return true }, func(status Status) { statuses = append(statuses, status) }, ) - _ = worker.Run( - withGeneration(context.Background(), 8), - FeedConfig{Domain: "/audio", UUID: "audio", Active: true}, - ) + config := FeedConfig{Domain: "/audio", UUID: "audio", Active: true} + _ = worker.Run(withGeneration(context.Background(), 8), config) if len(statuses) == 0 { t.Fatal("no statuses emitted") } @@ -138,6 +136,9 @@ func TestAudioWorkerStatusesInheritGeneration(t *testing.T) { if status.Generation != 8 { t.Fatalf("status generation = %d, want 8: %+v", status.Generation, status) } + if status.Feed != config { + t.Fatalf("status feed = %#v, want %#v", status.Feed, config) + } } } diff --git a/internal/playback/playlist_readiness.go b/internal/playback/playlist_readiness.go index 49aead4..201860c 100644 --- a/internal/playback/playlist_readiness.go +++ b/internal/playback/playlist_readiness.go @@ -18,6 +18,7 @@ func IsSessionPlaying( statuses.Video, statuses.HasVideo, session.Generation, + session.Plan.Video, ) { return false } @@ -25,17 +26,17 @@ func IsSessionPlaying( statuses.Audio, statuses.HasAudio, session.Generation, + session.Plan.Audio, ) { return false } return true case TopologySynchronized: - return statusIsPlaying( - statuses.Sync, - statuses.HasSync, - session.Generation, - ) + return statuses.HasSync && + statuses.Sync.Generation == session.Generation && + statuses.Sync.State == StatePlaying && + sameSyncSource(statuses.Sync.Pair, session.Plan.Sync) case TopologyIdle: return false @@ -45,8 +46,23 @@ func IsSessionPlaying( } } -func statusIsPlaying(status Status, present bool, generation uint64) bool { +func statusIsPlaying( + status Status, + present bool, + generation uint64, + feed FeedConfig, +) bool { return present && status.Generation == generation && - status.State == StatePlaying + status.State == StatePlaying && + sameFeedSource(status.Feed, feed) +} + +func sameFeedSource(a, b FeedConfig) bool { + return a.Domain == b.Domain && a.UUID == b.UUID +} + +func sameSyncSource(a, b SyncPairConfig) bool { + return sameFeedSource(a.Video, b.Video) && + sameFeedSource(a.Audio, b.Audio) } diff --git a/internal/playback/playlist_readiness_test.go b/internal/playback/playlist_readiness_test.go index 4102b47..2992b63 100644 --- a/internal/playback/playlist_readiness_test.go +++ b/internal/playback/playlist_readiness_test.go @@ -5,7 +5,14 @@ import "testing" func TestIsSessionPlaying(t *testing.T) { const generation = 4 playing := func(unit Unit) Status { - return Status{Unit: unit, State: StatePlaying, Generation: generation} + status := Status{Unit: unit, State: StatePlaying, Generation: generation} + switch unit { + case UnitVideo: + status.Feed = FeedConfig{UUID: "video"} + case UnitAudio: + status.Feed = FeedConfig{UUID: "audio"} + } + return status } tests := []struct { name string @@ -171,6 +178,72 @@ func TestIsSessionPlaying(t *testing.T) { HasSync: true, }, }, + { + name: "video status has wrong UUID", + session: SessionSnapshot{ + Generation: generation, + Plan: SessionPlan{ + Topology: TopologyIndependent, + Video: FeedConfig{Domain: "domain", UUID: "video", Active: true}, + }, + }, + statuses: PlaybackStatusSnapshot{ + Generation: generation, + Video: Status{ + Unit: UnitVideo, + State: StatePlaying, + Generation: generation, + Feed: FeedConfig{Domain: "domain", UUID: "other"}, + }, + HasVideo: true, + }, + }, + { + name: "audio status has wrong domain", + session: SessionSnapshot{ + Generation: generation, + Plan: SessionPlan{ + Topology: TopologyIndependent, + Audio: FeedConfig{Domain: "audio-domain", UUID: "audio", Active: true}, + }, + }, + statuses: PlaybackStatusSnapshot{ + Generation: generation, + Audio: Status{ + Unit: UnitAudio, + State: StatePlaying, + Generation: generation, + Feed: FeedConfig{Domain: "other-domain", UUID: "audio"}, + }, + HasAudio: true, + }, + }, + { + name: "sync status has wrong audio source", + session: SessionSnapshot{ + Generation: generation, + Plan: SessionPlan{ + Topology: TopologySynchronized, + Sync: SyncPairConfig{ + Video: FeedConfig{Domain: "domain", UUID: "video"}, + Audio: FeedConfig{Domain: "domain", UUID: "audio"}, + }, + }, + }, + statuses: PlaybackStatusSnapshot{ + Generation: generation, + Sync: Status{ + Unit: UnitSync, + State: StatePlaying, + Generation: generation, + Pair: SyncPairConfig{ + Video: FeedConfig{Domain: "domain", UUID: "video"}, + Audio: FeedConfig{Domain: "domain", UUID: "other-audio"}, + }, + }, + HasSync: true, + }, + }, { name: "idle", session: SessionSnapshot{ diff --git a/internal/playback/state.go b/internal/playback/state.go index 67c6fc9..92e91cd 100644 --- a/internal/playback/state.go +++ b/internal/playback/state.go @@ -26,9 +26,13 @@ const ( ) type Status struct { - Unit Unit - State State - Generation uint64 + Unit Unit + State State + Generation uint64 + + Feed FeedConfig // Video/Audio worker + Pair SyncPairConfig // Sync worker + Attempt int FailedAttempts int RetryIn time.Duration diff --git a/internal/playback/sync_worker.go b/internal/playback/sync_worker.go index 57dd6ef..04fdd64 100644 --- a/internal/playback/sync_worker.go +++ b/internal/playback/sync_worker.go @@ -59,8 +59,13 @@ func NewSyncWorker( }, nil } -func (w *SyncWorker) emit(ctx context.Context, status Status) { +func (w *SyncWorker) emit( + ctx context.Context, + pair SyncPairConfig, + status Status, +) { status.Generation = generationFromContext(ctx) + status.Pair = pair if w.observer != nil { w.observer(status) } @@ -80,6 +85,7 @@ func (w *SyncWorker) Run( if !videoConfig.Active || !audioConfig.Active { return ErrSyncFeedsInactive } + pair := SyncPairConfig{Video: videoConfig, Audio: audioConfig} attemptNumber := 0 var latestRetry retryEvent @@ -91,7 +97,7 @@ func (w *SyncWorker) Run( if attemptNumber > 1 { state = StateReconnecting } - w.emit(ctx, Status{ + w.emit(ctx, pair, Status{ Unit: UnitSync, State: state, Attempt: attemptNumber, @@ -100,7 +106,7 @@ func (w *SyncWorker) Run( attemptAudioSink := &stabilityAudioSink{ sink: w.audioSink, onStable: func() { - w.emit(ctx, Status{ + w.emit(ctx, pair, Status{ Unit: UnitSync, State: StatePlaying, Attempt: attemptNumber, @@ -134,7 +140,7 @@ func (w *SyncWorker) Run( if !event.WillRetry { return } - w.emit(ctx, Status{ + w.emit(ctx, pair, Status{ Unit: UnitSync, State: StateReconnecting, Attempt: attemptNumber + 1, @@ -152,18 +158,18 @@ func (w *SyncWorker) Run( observeRetry, ) if ctx.Err() != nil { - w.emit(ctx, Status{ + w.emit(ctx, pair, Status{ Unit: UnitSync, State: StateStopping, }) - w.emit(ctx, Status{ + w.emit(ctx, pair, Status{ Unit: UnitSync, State: StateIdle, }) return ctx.Err() } if err != nil { - w.emit(ctx, Status{ + w.emit(ctx, pair, Status{ Unit: UnitSync, State: StateFailed, Attempt: attemptNumber, @@ -172,7 +178,7 @@ func (w *SyncWorker) Run( }) return err } - w.emit(ctx, Status{ + w.emit(ctx, pair, Status{ Unit: UnitSync, State: StateIdle, }) diff --git a/internal/playback/sync_worker_test.go b/internal/playback/sync_worker_test.go index ecb20cb..af7c6dd 100644 --- a/internal/playback/sync_worker_test.go +++ b/internal/playback/sync_worker_test.go @@ -146,6 +146,10 @@ func TestSyncWorkerStatusesInheritGeneration(t *testing.T) { if status.Generation != 9 { t.Fatalf("status generation = %d, want 9: %+v", status.Generation, status) } + wantPair := SyncPairConfig{Video: video, Audio: audio} + if status.Pair != wantPair { + t.Fatalf("status pair = %#v, want %#v", status.Pair, wantPair) + } } } diff --git a/internal/playback/video_worker.go b/internal/playback/video_worker.go index 52305cd..7be9a85 100644 --- a/internal/playback/video_worker.go +++ b/internal/playback/video_worker.go @@ -72,8 +72,9 @@ func (s *stabilityVideoSink) ConsumeVideo( return err } -func (w *VideoWorker) emit(ctx context.Context, status Status) { +func (w *VideoWorker) emit(ctx context.Context, config FeedConfig, status Status) { status.Generation = generationFromContext(ctx) + status.Feed = config if w.observer != nil { w.observer(status) } @@ -100,7 +101,7 @@ func (w *VideoWorker) Run( if attemptNumber > 1 { state = StateReconnecting } - w.emit(ctx, Status{ + w.emit(ctx, config, Status{ Unit: UnitVideo, State: state, Attempt: attemptNumber, @@ -109,7 +110,7 @@ func (w *VideoWorker) Run( attemptSink := &stabilityVideoSink{ sink: w.sink, onStable: func() { - w.emit(ctx, Status{ + w.emit(ctx, config, Status{ Unit: UnitVideo, State: StatePlaying, Attempt: attemptNumber, @@ -135,7 +136,7 @@ func (w *VideoWorker) Run( return } - w.emit(ctx, Status{ + w.emit(ctx, config, Status{ Unit: UnitVideo, State: StateReconnecting, Attempt: attemptNumber + 1, @@ -155,11 +156,11 @@ func (w *VideoWorker) Run( ) if ctx.Err() != nil { - w.emit(ctx, Status{ + w.emit(ctx, config, Status{ Unit: UnitVideo, State: StateStopping, }) - w.emit(ctx, Status{ + w.emit(ctx, config, Status{ Unit: UnitVideo, State: StateIdle, }) @@ -167,7 +168,7 @@ func (w *VideoWorker) Run( } if err != nil { - w.emit(ctx, Status{ + w.emit(ctx, config, Status{ Unit: UnitVideo, State: StateFailed, Attempt: attemptNumber, @@ -177,7 +178,7 @@ func (w *VideoWorker) Run( return err } - w.emit(ctx, Status{ + w.emit(ctx, config, Status{ Unit: UnitVideo, State: StateIdle, }) diff --git a/internal/playback/video_worker_test.go b/internal/playback/video_worker_test.go index a5a9434..76a48c0 100644 --- a/internal/playback/video_worker_test.go +++ b/internal/playback/video_worker_test.go @@ -162,7 +162,8 @@ func TestVideoWorkerStatusesInheritGeneration(t *testing.T) { func(error) bool { return true }, func(status Status) { statuses = append(statuses, status) }, ) - _ = worker.Run(withGeneration(context.Background(), 7), activeVideoConfig()) + config := activeVideoConfig() + _ = worker.Run(withGeneration(context.Background(), 7), config) if len(statuses) == 0 { t.Fatal("no statuses emitted") } @@ -170,6 +171,9 @@ func TestVideoWorkerStatusesInheritGeneration(t *testing.T) { if status.Generation != 7 { t.Fatalf("status generation = %d, want 7: %+v", status.Generation, status) } + if status.Feed != config { + t.Fatalf("status feed = %#v, want %#v", status.Feed, config) + } } }