generation-aware playback statuses

cmd/mxl-player/main.go

@@ -99,6 +99,8 @@ func main() {
 	// timelapse audio: 5fbec3b1-1b0f-417d-9059-8b94a47197ec
 	// f1 video: 5fbec3b1-1b0f-417d-9059-8b94a47197ef
 	// f1 audio: 5fbec3b1-1b0f-417d-9059-8b94a47197eb
+	// sync video: 2618979d-76a5-45e0-83cb-0f192978d1cd
+	// sync audio: 9d2a041b-01cf-4ee4-bffa-188fe093c99b
 	var args appArgs
 	flagSet := pflag.NewFlagSet(APP_NAME, pflag.ContinueOnError)
 	flagSet.SortFlags = false
@@ -200,6 +202,19 @@ func main() {
 	// ImGui init
 	gui := imgui.New()
 	defer gui.Destroy()
+	fontConfig := cimgui.NewFontConfig()
+	font := gui.IO().Fonts().AddFontFromFileTTFV(
+		"/home/itten/Downloads/JetBrainsMono/JetBrainsMonoNLNerdFontMono-Regular.ttf",
+		18,
+		fontConfig,
+		nil,
+	)
+	fontConfig.Destroy()
+	if font == nil || font.CData == nil {
+		log.Fatal("failed to load ImGui font")
+	}
+	gui.IO().SetFontDefault(font)
+	// sdl keys handler
 	sdl.StartTextInput(windowHandler)
 	defer sdl.StopTextInput(windowHandler)
 	// fin on ImGui init
@@ -421,6 +436,12 @@ func main() {
 	)
 	lastFrame = time.Now()
 
+	// ImGui
+	var (
+		settingWindowWidth  float32 = 700
+		settingsWindowState bool    = true
+	)
+
 	for running {
 		frameStart := time.Now()
 		var event [128]byte
@@ -447,6 +468,8 @@ func main() {
 					resized = true
 				case sdl.KeyF1:
 					showStats = !showStats
+				case sdl.KeyF2:
+					settingsWindowState = !settingsWindowState
 				}
 			}
 			gui.ProcessEvent(&event)
@@ -523,8 +546,6 @@ func main() {
 		// end of stats
 		if r != nil {
 			gui.BeginFrame(time.Since(lastFrame), int32(r.Extent().Width), int32(r.Extent().Height))
-			// test widget
-			// cimgui.Begin("Test")
 			if showStats {
 				cimgui.SetNextWindowPos(cimgui.Vec2{X: 10, Y: 10})
 				cimgui.SetNextWindowSize(cimgui.Vec2{X: 200, Y: 200})
@@ -540,13 +561,9 @@ func main() {
 						displayedVideoHeight,
 					))
 				}
-				cimgui.Text("\nPress F1 to hide stats")
-				cimgui.Text("Q or Esc to quit")
-				cimgui.Text("F for fullscreen")
 				cimgui.End()
 			}
-			cimgui.Begin("Connection")
+			// settings & info window
-
 			snapshot, hasSnapshot := player.Controller.Snapshot()
 			videoConfigured := videoStr != ""
 			audioConfigured := audioStr != ""
@@ -556,20 +573,63 @@ func main() {
 				videoConfigured = snapshot.Desired.Video.IsConfigured()
 				audioConfigured = snapshot.Desired.Audio.IsConfigured()
 				syncRequested = snapshot.Desired.SyncRequested
-				cimgui.Text(fmt.Sprintf(
-					"Topology: %s (generation %d)",
-					snapshot.Plan.Topology,
-					snapshot.Generation,
-				))
-			} else {
-				cimgui.Text("Topology: starting")
 			}
 
-			cimgui.Separator()
+			drawSettingsContents := func() {
+				var collapsingHeaderFlags cimgui.TreeNodeFlags = cimgui.TreeNodeFlagsDefaultOpen
+				drawFeedsSections := func() {
+					cimgui.SeparatorText("Video")
 					cimgui.InputTextWithHint("Video domain", "/dev/shm/mxl", &videoDomainStr, 0, nil)
 					cimgui.InputTextWithHint("Video UUID", "", &videoStr, 0, nil)
+					if videoActive {
+						cimgui.SameLine()
+						if cimgui.Button("Stop##video") {
+							videoActive = false
+							enqueueCommand(playback.SessionCommand{Kind: playback.CommandStopVideo})
+						}
+					}
+					if !videoActive && videoConfigured {
+						cimgui.SameLine()
+						if cimgui.Button("Resume##video") {
+							videoActive = true
+							enqueueCommand(playback.SessionCommand{Kind: playback.CommandResumeVideo})
+						}
+					}
+					if videoConfigured {
+						cimgui.SameLine()
+						if cimgui.Button("Remove##video") {
+							videoActive = false
+							videoStr = ""
+							enqueueCommand(playback.SessionCommand{Kind: playback.CommandRemoveVideo})
+						}
+					}
+					cimgui.SeparatorText("Audio")
 					cimgui.InputTextWithHint("Audio domain", "/dev/shm/mxl", &audioDomainStr, 0, nil)
 					cimgui.InputTextWithHint("Audio UUID", "", &audioStr, 0, nil)
+					if audioActive {
+						cimgui.SameLine()
+						if cimgui.Button("Stop##audio") {
+							audioActive = false
+							enqueueCommand(playback.SessionCommand{Kind: playback.CommandStopAudio})
+						}
+					}
+
+					if !audioActive && audioConfigured {
+						cimgui.SameLine()
+						if cimgui.Button("Resume##audio") {
+							audioActive = true
+							enqueueCommand(playback.SessionCommand{Kind: playback.CommandResumeAudio})
+						}
+					}
+					if audioConfigured {
+						cimgui.SameLine()
+						if cimgui.Button("Remove##audio") {
+							audioActive = false
+							audioStr = ""
+							enqueueCommand(playback.SessionCommand{Kind: playback.CommandRemoveAudio})
+						}
+					}
+					cimgui.SeparatorText("Controls")
 					if cimgui.Button("Apply feeds") {
 						doReconnect()
 					}
@@ -587,7 +647,7 @@ func main() {
 							enqueueCommand(playback.SessionCommand{Kind: playback.CommandResumeAll})
 						}
 					}
-
+					cimgui.SameLine()
 					if cimgui.Checkbox("Synchronize", &syncRequested) {
 						kind := playback.CommandDisableSync
 						if syncRequested {
@@ -595,6 +655,33 @@ func main() {
 						}
 						enqueueCommand(playback.SessionCommand{Kind: kind})
 					}
+					cimgui.SeparatorText("Feeds stats")
+					cimgui.Checkbox("Show stats", &showStats)
+				}
+				if cimgui.CollapsingHeaderTreeNodeFlagsV("Feeds", collapsingHeaderFlags) {
+					drawFeedsSections()
+				}
+
+				drawHotkeysSection := func() {
+					cimgui.Text("F1 - show/hide stats")
+					cimgui.Text("F2 - show/hide settings")
+					cimgui.Text("F - toggle fullscreen")
+					cimgui.Text("Q or Esc - quit")
+				}
+				if cimgui.CollapsingHeaderTreeNodeFlagsV("Hotkeys", collapsingHeaderFlags) {
+					drawHotkeysSection()
+				}
+
+				drawDebugSection := func() {
+					if hasSnapshot {
+						cimgui.Text(fmt.Sprintf(
+							"Topology: %s (generation %d)",
+							snapshot.Plan.Topology,
+							snapshot.Generation,
+						))
+					} else {
+						cimgui.Text("Topology: starting")
+					}
 					if hasSnapshot && syncRequested && snapshot.Plan.Topology != playback.TopologySynchronized {
 						switch {
 						case !videoConfigured || !audioConfigured:
@@ -607,72 +694,6 @@ func main() {
 							cimgui.TextWrapped("Sync requested but currently unavailable. Playing independently.")
 						}
 					}
-
-			cimgui.Separator()
-			cimgui.Text("Video")
-			if videoActive {
-				if cimgui.Button("Stop video") {
-					videoActive = false
-					enqueueCommand(playback.SessionCommand{Kind: playback.CommandStopVideo})
-				}
-			}
-			if !videoActive && videoConfigured {
-				cimgui.SameLine()
-				if cimgui.Button("Resume video") {
-					videoActive = true
-					enqueueCommand(playback.SessionCommand{Kind: playback.CommandResumeVideo})
-				}
-			}
-			if videoConfigured {
-				cimgui.SameLine()
-				if cimgui.Button("Remove video") {
-					videoActive = false
-					videoStr = ""
-					enqueueCommand(playback.SessionCommand{Kind: playback.CommandRemoveVideo})
-				}
-			}
-			if videoActive {
-				cimgui.Text("Video desired: active")
-			} else if videoConfigured {
-				cimgui.Text("Video desired: stopped")
-			} else {
-				cimgui.Text("Video desired: not configured")
-			}
-
-			cimgui.Separator()
-			cimgui.Text("Audio")
-			if audioActive {
-				if cimgui.Button("Stop audio") {
-					audioActive = false
-					enqueueCommand(playback.SessionCommand{Kind: playback.CommandStopAudio})
-				}
-			}
-
-			if !audioActive && audioConfigured {
-				cimgui.SameLine()
-				if cimgui.Button("Resume audio") {
-					audioActive = true
-					enqueueCommand(playback.SessionCommand{Kind: playback.CommandResumeAudio})
-				}
-			}
-
-			if audioConfigured {
-				cimgui.SameLine()
-				if cimgui.Button("Remove audio") {
-					audioActive = false
-					audioStr = ""
-					enqueueCommand(playback.SessionCommand{Kind: playback.CommandRemoveAudio})
-				}
-			}
-
-			if audioActive {
-				cimgui.Text("Audio desired: active")
-			} else if audioConfigured {
-				cimgui.Text("Audio desired: stopped")
-			} else {
-				cimgui.Text("Audio desired: not configured")
-			}
-
 					cimgui.Separator()
 					cimgui.Text("Current playback")
 					if hasSnapshot {
@@ -692,11 +713,36 @@ func main() {
 					} else {
 						cimgui.Text("Playback controller is starting")
 					}
-
 					cimgui.Separator()
-			cimgui.Checkbox("Show stats", &showStats)
+					if videoActive {
+						cimgui.Text("Video desired: active")
+					} else if videoConfigured {
+						cimgui.Text("Video desired: stopped")
+					} else {
+						cimgui.Text("Video desired: not configured")
+					}
+					if audioActive {
+						cimgui.Text("Audio desired: active")
+					} else if audioConfigured {
+						cimgui.Text("Audio desired: stopped")
+					} else {
+						cimgui.Text("Audio desired: not configured")
+					}
 
+				}
+				if cimgui.CollapsingHeaderTreeNodeFlagsV("Debug Info", collapsingHeaderFlags) {
+					drawDebugSection()
+				}
+			}
+
+			if settingsWindowState {
+				cimgui.SetNextWindowPos(cimgui.Vec2{X: float32(r.Extent().Width) - settingWindowWidth, Y: 0})
+				cimgui.SetNextWindowSize(cimgui.Vec2{X: settingWindowWidth, Y: float32(r.Extent().Height)})
+				if cimgui.BeginV("Settings & Info", &settingsWindowState, cimgui.WindowFlagsNone) {
+					drawSettingsContents()
+				}
 				cimgui.End()
+			}
 			gui.EndFrame()
 			lastFrame = time.Now()
 			// end of test widget

imgui.ini

@@ -1,6 +1,6 @@
 [Window][Debug##Default]
-Pos=60,60
+Pos=519,181
-Size=400,400
+Size=400,398
 Collapsed=0
 
 [Window][Test]
@@ -14,7 +14,27 @@ Size=200,200
 Collapsed=0
 
 [Window][Connection]
-Pos=322,130
+Pos=42,264
-Size=661,444
+Size=605,416
+Collapsed=0
+
+[Window][Test slider]
+Pos=1370,0
+Size=550,1080
+Collapsed=0
+
+[Window][Settings]
+Pos=730,0
+Size=550,720
+Collapsed=0
+
+[Window][Settings & Info]
+Pos=1220,0
+Size=700,1080
+Collapsed=0
+
+[Window][Seetings & Info]
+Pos=1220,0
+Size=700,1080
 Collapsed=0
 

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{

internal/sdl/sdl.go

@@ -32,6 +32,7 @@ const (
 	KeyF      uint32 = 0x66
 	KeyQ      uint32 = 0x71
 	KeyF1     uint32 = 0x4000003A
+	KeyF2     uint32 = 0x4000003B
 
 	InitAudio                  uint32 = 0x00000010
 	AudioDeviceDefaultPlayback uint32 = 0xFFFFFFFF

mxl-gst-scripts/loop-feed3.sh

@@ -0,0 +1,12 @@
+#!/bin/bash
+VIDEO_ID="2618979d-76a5-45e0-83cb-0f192978d1cd"
+AUDIO_ID="9d2a041b-01cf-4ee4-bffa-188fe093c99b"
+VIDEO_URI=$1
+if [[ -z "${VIDEO_URI}" ]] then
+    VIDEO_URI="${HOME}/Videos/test-vid/motogp.ts"
+fi
+export GST_PLUGIN_PATH="${HOME}/.gst-plugin:${GST_PLUGIN_PATH}"
+mxl-gst-looping-filesrc -d /dev/shm/mxl -i "${VIDEO_URI}" --video-id "${VIDEO_ID}" --audio-id "${AUDIO_ID}" 2>/dev/null
+# sleep 5
+# kill -9 $(pidof "mxl-gst-looping-filesrc")
+# echo -e "\ngst-looping-filesrc killed"