diff --git a/internal/playback/audio_worker.go b/internal/playback/audio_worker.go index 066db14..e0ab115 100644 --- a/internal/playback/audio_worker.go +++ b/internal/playback/audio_worker.go @@ -72,7 +72,8 @@ func (s *stabilityAudioSink) ConsumeAudio( return err } -func (w *AudioWorker) emit(status Status) { +func (w *AudioWorker) emit(ctx context.Context, status Status) { + status.Generation = generationFromContext(ctx) if w.observer != nil { w.observer(status) } @@ -99,7 +100,7 @@ func (w *AudioWorker) Run( if attemptNumber > 1 { state = StateReconnecting } - w.emit(Status{ + w.emit(ctx, Status{ Unit: UnitAudio, State: state, Attempt: attemptNumber, @@ -108,7 +109,7 @@ func (w *AudioWorker) Run( attemptSink := &stabilityAudioSink{ sink: w.sink, onStable: func() { - w.emit(Status{ + w.emit(ctx, Status{ Unit: UnitAudio, State: StatePlaying, Attempt: attemptNumber, @@ -134,7 +135,7 @@ func (w *AudioWorker) Run( return } - w.emit(Status{ + w.emit(ctx, Status{ Unit: UnitAudio, State: StateReconnecting, Attempt: attemptNumber + 1, @@ -154,11 +155,11 @@ func (w *AudioWorker) Run( ) if ctx.Err() != nil { - w.emit(Status{ + w.emit(ctx, Status{ Unit: UnitAudio, State: StateStopping, }) - w.emit(Status{ + w.emit(ctx, Status{ Unit: UnitAudio, State: StateIdle, }) @@ -166,7 +167,7 @@ func (w *AudioWorker) Run( } if err != nil { - w.emit(Status{ + w.emit(ctx, Status{ Unit: UnitAudio, State: StateFailed, Attempt: attemptNumber, @@ -176,7 +177,7 @@ func (w *AudioWorker) Run( return err } - w.emit(Status{ + w.emit(ctx, Status{ Unit: UnitAudio, State: StateIdle, }) diff --git a/internal/playback/audio_worker_test.go b/internal/playback/audio_worker_test.go index 82becef..a560ba1 100644 --- a/internal/playback/audio_worker_test.go +++ b/internal/playback/audio_worker_test.go @@ -116,6 +116,31 @@ func TestAudioWorkerRejectsInactiveFeed(t *testing.T) { } } +func TestAudioWorkerStatusesInheritGeneration(t *testing.T) { + openErr := errors.New("unavailable") + var statuses []Status + worker := newAudioWorkerForTest( + t, + &queuedAudioFactory{errs: []error{openErr}}, + &fakeAudioSink{}, + 1, + 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}, + ) + if len(statuses) == 0 { + t.Fatal("no statuses emitted") + } + for _, status := range statuses { + if status.Generation != 8 { + t.Fatalf("status generation = %d, want 8: %+v", status.Generation, status) + } + } +} + func TestAudioWorkerPublishesPlayingThenFailed(t *testing.T) { readErr := errors.New("audio disappeared") reader := &fakeAudioReader{ diff --git a/internal/playback/session_controller.go b/internal/playback/session_controller.go index 85879c1..85fb2b4 100644 --- a/internal/playback/session_controller.go +++ b/internal/playback/session_controller.go @@ -109,9 +109,8 @@ func (c *SessionController) Run( } desired := initial - runtime := c.startSessionRuntime(ctx, plan) - generation := uint64(1) + runtime := c.startSessionRuntime(ctx, plan, generation) c.publish(SessionSnapshot{ Desired: initial, Plan: plan, @@ -148,11 +147,16 @@ func (c *SessionController) Run( continue } + nextGeneration := generation + if plan.Topology != nextPlan.Topology { + nextGeneration++ + } nextRuntime, err := c.reconcileSessionRuntime( ctx, runtime, plan, nextPlan, + nextGeneration, ) if err != nil { stopSessionRuntime(runtime) @@ -162,9 +166,7 @@ func (c *SessionController) Run( return err } - if plan.Topology != nextPlan.Topology { - generation++ - } + generation = nextGeneration desired = nextDesired plan = nextPlan runtime = nextRuntime @@ -180,12 +182,13 @@ func (c *SessionController) Run( func (c *SessionController) startSessionRuntime( ctx context.Context, plan SessionPlan, + generation uint64, ) *sessionRuntime { if plan.Topology == TopologyIdle { return &sessionRuntime{topology: TopologyIdle} } - runtimeCtx, cancel := context.WithCancel(ctx) + runtimeCtx, cancel := context.WithCancel(withGeneration(ctx, generation)) runtime := &sessionRuntime{ topology: plan.Topology, cancel: cancel, @@ -255,13 +258,14 @@ func (c *SessionController) reconcileSessionRuntime( runtime *sessionRuntime, current SessionPlan, next SessionPlan, + nextGeneration uint64, ) (*sessionRuntime, error) { if current.Topology != next.Topology { stopSessionRuntime(runtime) if err := ctx.Err(); err != nil { return runtime, err } - return c.startSessionRuntime(ctx, next), nil + return c.startSessionRuntime(ctx, next, nextGeneration), nil } switch next.Topology { diff --git a/internal/playback/session_controller_test.go b/internal/playback/session_controller_test.go index 89f2bf6..f0468fc 100644 --- a/internal/playback/session_controller_test.go +++ b/internal/playback/session_controller_test.go @@ -88,10 +88,11 @@ func TestNewSessionControllerStoresSyncPredicate(t *testing.T) { } type controllerEvent struct { - unit Unit - action string - feed FeedConfig - pair SyncPairConfig + unit Unit + action string + generation uint64 + feed FeedConfig + pair SyncPairConfig } type recordingVideoSlot struct{ events chan<- controllerEvent } @@ -101,7 +102,10 @@ func (s recordingVideoSlot) Run( initial FeedConfig, commands <-chan FeedConfig, ) error { - s.events <- controllerEvent{unit: UnitVideo, action: "start", feed: initial} + s.events <- controllerEvent{ + unit: UnitVideo, action: "start", + generation: generationFromContext(ctx), feed: initial, + } for { select { case config := <-commands: @@ -120,7 +124,10 @@ func (s recordingAudioSlot) Run( initial FeedConfig, commands <-chan FeedConfig, ) error { - s.events <- controllerEvent{unit: UnitAudio, action: "start", feed: initial} + s.events <- controllerEvent{ + unit: UnitAudio, action: "start", + generation: generationFromContext(ctx), feed: initial, + } for { select { case config := <-commands: @@ -139,7 +146,10 @@ func (s recordingSyncSlot) Run( initial SyncPairConfig, commands <-chan SyncPairConfig, ) error { - s.events <- controllerEvent{unit: UnitSync, action: "start", pair: initial} + s.events <- controllerEvent{ + unit: UnitSync, action: "start", + generation: generationFromContext(ctx), pair: initial, + } for { select { case config := <-commands: @@ -291,6 +301,9 @@ func TestSessionControllerStopsIndependentSlotsBeforeStartingSync(t *testing.T) if !stopped[UnitVideo] || !stopped[UnitAudio] { t.Fatalf("sync started before both independent slots stopped: %v", stopped) } + if event.generation != 2 { + t.Fatalf("sync runtime generation = %d, want 2", event.generation) + } break } if event.action != "stop" || (event.unit != UnitVideo && event.unit != UnitAudio) { diff --git a/internal/playback/state.go b/internal/playback/state.go index 1d07333..67c6fc9 100644 --- a/internal/playback/state.go +++ b/internal/playback/state.go @@ -1,6 +1,7 @@ package playback import ( + "context" "fmt" "time" ) @@ -27,6 +28,7 @@ const ( type Status struct { Unit Unit State State + Generation uint64 Attempt int FailedAttempts int RetryIn time.Duration @@ -35,6 +37,17 @@ type Status struct { type StatusObserver func(Status) +type generationContextKey struct{} + +func withGeneration(ctx context.Context, generation uint64) context.Context { + return context.WithValue(ctx, generationContextKey{}, generation) +} + +func generationFromContext(ctx context.Context) uint64 { + generation, _ := ctx.Value(generationContextKey{}).(uint64) + return generation +} + func (u Unit) String() string { switch u { case UnitVideo: diff --git a/internal/playback/state_test.go b/internal/playback/state_test.go index 98892fa..2f4093c 100644 --- a/internal/playback/state_test.go +++ b/internal/playback/state_test.go @@ -1,11 +1,22 @@ package playback import ( + "context" "errors" "testing" "time" ) +func TestGenerationContext(t *testing.T) { + if got := generationFromContext(context.Background()); got != 0 { + t.Fatalf("background generation = %d, want 0", got) + } + ctx := withGeneration(context.Background(), 42) + if got := generationFromContext(ctx); got != 42 { + t.Fatalf("generation = %d, want 42", got) + } +} + func TestStatusPreservesValues(t *testing.T) { wantErr := errors.New("producer missing") status := Status{ diff --git a/internal/playback/status_store.go b/internal/playback/status_store.go index 89a20b8..d9ce79f 100644 --- a/internal/playback/status_store.go +++ b/internal/playback/status_store.go @@ -3,8 +3,9 @@ package playback import "sync" type StatusStore struct { - mu sync.RWMutex - statuses map[Unit]Status + mu sync.RWMutex + generation uint64 + statuses map[Unit]Status } func NewStatusStore() *StatusStore { @@ -15,8 +16,15 @@ func NewStatusStore() *StatusStore { func (s *StatusStore) Observe(status Status) { s.mu.Lock() + defer s.mu.Unlock() + if status.Generation < s.generation { + return + } + if status.Generation > s.generation { + clear(s.statuses) + s.generation = status.Generation + } s.statuses[status.Unit] = status - s.mu.Unlock() } func (s *StatusStore) Snapshot(unit Unit) (Status, bool) { diff --git a/internal/playback/status_store_test.go b/internal/playback/status_store_test.go index 8aa308c..fb3a25e 100644 --- a/internal/playback/status_store_test.go +++ b/internal/playback/status_store_test.go @@ -106,3 +106,50 @@ func TestStatusStoreConcurrentAccess(t *testing.T) { } } } + +func TestStatusStoreNewGenerationClearsPreviousUnits(t *testing.T) { + store := NewStatusStore() + store.Observe(Status{Unit: UnitVideo, State: StatePlaying, Generation: 1}) + store.Observe(Status{Unit: UnitAudio, State: StatePlaying, Generation: 1}) + want := Status{Unit: UnitSync, State: StateConnecting, Generation: 2} + store.Observe(want) + + if _, ok := store.Snapshot(UnitVideo); ok { + t.Fatal("video status survived generation change") + } + if _, ok := store.Snapshot(UnitAudio); ok { + t.Fatal("audio status survived generation change") + } + if got, ok := store.Snapshot(UnitSync); !ok || got != want { + t.Fatalf("sync Snapshot() = %#v, %t; want %#v, true", got, ok, want) + } +} + +func TestStatusStoreIgnoresOlderGeneration(t *testing.T) { + store := NewStatusStore() + want := Status{Unit: UnitSync, State: StatePlaying, Generation: 3} + store.Observe(want) + store.Observe(Status{Unit: UnitVideo, State: StateIdle, Generation: 2}) + + if _, ok := store.Snapshot(UnitVideo); ok { + t.Fatal("older video status was stored") + } + if got, ok := store.Snapshot(UnitSync); !ok || got != want { + t.Fatalf("sync Snapshot() = %#v, %t; want %#v, true", got, ok, want) + } +} + +func TestStatusStoreKeepsEqualGenerationUnitsIndependent(t *testing.T) { + store := NewStatusStore() + wantVideo := Status{Unit: UnitVideo, State: StatePlaying, Generation: 4} + wantAudio := Status{Unit: UnitAudio, State: StateReconnecting, Generation: 4} + store.Observe(wantVideo) + store.Observe(wantAudio) + + if got, ok := store.Snapshot(UnitVideo); !ok || got != wantVideo { + t.Fatalf("video Snapshot() = %#v, %t", got, ok) + } + if got, ok := store.Snapshot(UnitAudio); !ok || got != wantAudio { + t.Fatalf("audio Snapshot() = %#v, %t", got, ok) + } +} diff --git a/internal/playback/sync_worker.go b/internal/playback/sync_worker.go index f058e98..57dd6ef 100644 --- a/internal/playback/sync_worker.go +++ b/internal/playback/sync_worker.go @@ -59,7 +59,8 @@ func NewSyncWorker( }, nil } -func (w *SyncWorker) emit(status Status) { +func (w *SyncWorker) emit(ctx context.Context, status Status) { + status.Generation = generationFromContext(ctx) if w.observer != nil { w.observer(status) } @@ -90,7 +91,7 @@ func (w *SyncWorker) Run( if attemptNumber > 1 { state = StateReconnecting } - w.emit(Status{ + w.emit(ctx, Status{ Unit: UnitSync, State: state, Attempt: attemptNumber, @@ -99,7 +100,7 @@ func (w *SyncWorker) Run( attemptAudioSink := &stabilityAudioSink{ sink: w.audioSink, onStable: func() { - w.emit(Status{ + w.emit(ctx, Status{ Unit: UnitSync, State: StatePlaying, Attempt: attemptNumber, @@ -133,7 +134,7 @@ func (w *SyncWorker) Run( if !event.WillRetry { return } - w.emit(Status{ + w.emit(ctx, Status{ Unit: UnitSync, State: StateReconnecting, Attempt: attemptNumber + 1, @@ -151,18 +152,18 @@ func (w *SyncWorker) Run( observeRetry, ) if ctx.Err() != nil { - w.emit(Status{ + w.emit(ctx, Status{ Unit: UnitSync, State: StateStopping, }) - w.emit(Status{ + w.emit(ctx, Status{ Unit: UnitSync, State: StateIdle, }) return ctx.Err() } if err != nil { - w.emit(Status{ + w.emit(ctx, Status{ Unit: UnitSync, State: StateFailed, Attempt: attemptNumber, @@ -171,7 +172,7 @@ func (w *SyncWorker) Run( }) return err } - w.emit(Status{ + w.emit(ctx, Status{ Unit: UnitSync, State: StateIdle, }) diff --git a/internal/playback/sync_worker_test.go b/internal/playback/sync_worker_test.go index b7ff47d..ecb20cb 100644 --- a/internal/playback/sync_worker_test.go +++ b/internal/playback/sync_worker_test.go @@ -126,6 +126,29 @@ func TestSyncWorkerRejectsInvalidOrInactiveFeeds(t *testing.T) { } } +func TestSyncWorkerStatusesInheritGeneration(t *testing.T) { + openErr := errors.New("unavailable") + var statuses []Status + worker := newTestSyncWorker( + t, + &scriptedSyncFactory{results: []syncOpenResult{{err: openErr}}}, + &fakeVideoSink{}, + &fakeAudioSink{}, + 1, + func(status Status) { statuses = append(statuses, status) }, + ) + video, audio := activeSyncConfigs() + _ = worker.Run(withGeneration(context.Background(), 9), video, audio) + if len(statuses) == 0 { + t.Fatal("no statuses emitted") + } + for _, status := range statuses { + if status.Generation != 9 { + t.Fatalf("status generation = %d, want 9: %+v", status.Generation, status) + } + } +} + func TestSyncWorkerExhaustsOpenRetries(t *testing.T) { openErr := errors.New("sync producer unavailable") factory := &scriptedSyncFactory{results: []syncOpenResult{{err: openErr}, {err: openErr}}} diff --git a/internal/playback/video_worker.go b/internal/playback/video_worker.go index bc929a6..52305cd 100644 --- a/internal/playback/video_worker.go +++ b/internal/playback/video_worker.go @@ -72,7 +72,8 @@ func (s *stabilityVideoSink) ConsumeVideo( return err } -func (w *VideoWorker) emit(status Status) { +func (w *VideoWorker) emit(ctx context.Context, status Status) { + status.Generation = generationFromContext(ctx) if w.observer != nil { w.observer(status) } @@ -99,7 +100,7 @@ func (w *VideoWorker) Run( if attemptNumber > 1 { state = StateReconnecting } - w.emit(Status{ + w.emit(ctx, Status{ Unit: UnitVideo, State: state, Attempt: attemptNumber, @@ -108,7 +109,7 @@ func (w *VideoWorker) Run( attemptSink := &stabilityVideoSink{ sink: w.sink, onStable: func() { - w.emit(Status{ + w.emit(ctx, Status{ Unit: UnitVideo, State: StatePlaying, Attempt: attemptNumber, @@ -134,7 +135,7 @@ func (w *VideoWorker) Run( return } - w.emit(Status{ + w.emit(ctx, Status{ Unit: UnitVideo, State: StateReconnecting, Attempt: attemptNumber + 1, @@ -154,11 +155,11 @@ func (w *VideoWorker) Run( ) if ctx.Err() != nil { - w.emit(Status{ + w.emit(ctx, Status{ Unit: UnitVideo, State: StateStopping, }) - w.emit(Status{ + w.emit(ctx, Status{ Unit: UnitVideo, State: StateIdle, }) @@ -166,7 +167,7 @@ func (w *VideoWorker) Run( } if err != nil { - w.emit(Status{ + w.emit(ctx, Status{ Unit: UnitVideo, State: StateFailed, Attempt: attemptNumber, @@ -176,7 +177,7 @@ func (w *VideoWorker) Run( return err } - w.emit(Status{ + w.emit(ctx, Status{ Unit: UnitVideo, State: StateIdle, }) diff --git a/internal/playback/video_worker_test.go b/internal/playback/video_worker_test.go index 30a96b9..a5a9434 100644 --- a/internal/playback/video_worker_test.go +++ b/internal/playback/video_worker_test.go @@ -151,6 +151,28 @@ func TestVideoWorkerRejectsInactiveFeed(t *testing.T) { } } +func TestVideoWorkerStatusesInheritGeneration(t *testing.T) { + openErr := errors.New("unavailable") + var statuses []Status + worker := newTestVideoWorker( + t, + &scriptedVideoFactory{results: []videoOpenResult{{err: openErr}}}, + &fakeVideoSink{}, + 1, + func(error) bool { return true }, + func(status Status) { statuses = append(statuses, status) }, + ) + _ = worker.Run(withGeneration(context.Background(), 7), activeVideoConfig()) + if len(statuses) == 0 { + t.Fatal("no statuses emitted") + } + for _, status := range statuses { + if status.Generation != 7 { + t.Fatalf("status generation = %d, want 7: %+v", status.Generation, status) + } + } +} + func TestVideoWorkerExhaustsOpenRetries(t *testing.T) { openErr := errors.New("producer unavailable") factory := &scriptedVideoFactory{