generation-aware playback statuses

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,
 	})

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{

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 = nextGeneration
-				generation++
-			}
 			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 {

internal/playback/session_controller_test.go

@@ -90,6 +90,7 @@ func TestNewSessionControllerStoresSyncPredicate(t *testing.T) {
 type controllerEvent struct {
 	unit       Unit
 	action     string
+	generation uint64
 	feed       FeedConfig
 	pair       SyncPairConfig
 }
@@ -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) {

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:

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{

internal/playback/status_store.go

@@ -4,6 +4,7 @@ import "sync"
 
 type StatusStore struct {
 	mu         sync.RWMutex
+	generation uint64
 	statuses   map[Unit]Status
 }
 
@@ -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) {

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)
+	}
+}

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,
 	})

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}}}

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,
 	})

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{