internal/playback/session_controller.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"sync"
 )
 
 type VideoSlotRunner interface {
@@ -40,11 +41,21 @@ var (
 	ErrSyncSlotRequired  = errors.New("sync slot is required")
 )
 
+type SessionSnapshot struct {
+	Desired    SessionConfig
+	Plan       SessionPlan
+	Generation uint64
+}
+
 type SessionController struct {
 	videoSlot VideoSlotRunner
 	audioSlot AudioSlotRunner
 	syncSlot  SyncSlotRunner
 	canSync   SyncPredicate
+
+	mu          sync.RWMutex
+	snapshot    SessionSnapshot
+	hasSnapshot bool
 }
 
 func NewSessionController(
@@ -100,6 +111,13 @@ func (c *SessionController) Run(
 	desired := initial
 	runtime := c.startSessionRuntime(ctx, plan)
 
+	generation := uint64(1)
+	c.publish(SessionSnapshot{
+		Desired:    initial,
+		Plan:       plan,
+		Generation: generation,
+	})
+
 	for {
 		var runtimeDone <-chan struct{}
 		if runtime != nil {
@@ -144,9 +162,17 @@ func (c *SessionController) Run(
 				return err
 			}
 
+			if plan.Topology != nextPlan.Topology {
+				generation++
+			}
 			desired = nextDesired
 			plan = nextPlan
 			runtime = nextRuntime
+			c.publish(SessionSnapshot{
+				Desired:    desired,
+				Plan:       plan,
+				Generation: generation,
+			})
 		}
 	}
 }
@@ -301,3 +327,18 @@ func sessionRuntimeSendError(ctx context.Context, runtime *sessionRuntime) error
 	<-runtime.done
 	return unexpectedSessionRuntimeError(runtime.result)
 }
+
+func (c *SessionController) Snapshot() (SessionSnapshot, bool) {
+	c.mu.RLock()
+	snapshot := c.snapshot
+	ok := c.hasSnapshot
+	c.mu.RUnlock()
+	return snapshot, ok
+}
+
+func (c *SessionController) publish(snapshot SessionSnapshot) {
+	c.mu.Lock()
+	c.snapshot = snapshot
+	c.hasSnapshot = true
+	c.mu.Unlock()
+}

internal/playback/session_controller_test.go

@@ -3,6 +3,7 @@ package playback
 import (
 	"context"
 	"errors"
+	"sync"
 	"testing"
 	"time"
 )
@@ -187,6 +188,25 @@ func receiveIndependentStarts(t *testing.T, events <-chan controllerEvent) {
 	}
 }
 
+func waitControllerSnapshot(
+	t *testing.T,
+	controller *SessionController,
+	match func(SessionSnapshot) bool,
+) SessionSnapshot {
+	t.Helper()
+	deadline := time.Now().Add(time.Second)
+	for {
+		if snapshot, ok := controller.Snapshot(); ok && match(snapshot) {
+			return snapshot
+		}
+		if time.Now().After(deadline) {
+			snapshot, ok := controller.Snapshot()
+			t.Fatalf("snapshot timed out: %#v, available=%t", snapshot, ok)
+		}
+		time.Sleep(time.Millisecond)
+	}
+}
+
 func TestSessionControllerUpdatesOnlyChangedIndependentSlot(t *testing.T) {
 	events := make(chan controllerEvent, 32)
 	controller := newRecordingController(t, events)
@@ -277,3 +297,126 @@ func TestSessionControllerCancellationStopsAndJoinsRuntime(t *testing.T) {
 		t.Fatal("Run() did not stop after cancellation")
 	}
 }
+
+func TestSessionControllerSnapshotTracksDesiredPlanAndGeneration(t *testing.T) {
+	events := make(chan controllerEvent, 64)
+	controller := newRecordingController(t, events)
+	if snapshot, ok := controller.Snapshot(); ok {
+		t.Fatalf("Snapshot() before Run = %#v, true; want unavailable", snapshot)
+	}
+
+	initial := validCommandSession()
+	initial.SyncRequested = false
+	commands := make(chan SessionCommand)
+	done := make(chan error, 1)
+	go func() { done <- controller.Run(context.Background(), initial, commands) }()
+	receiveIndependentStarts(t, events)
+
+	snapshot := waitControllerSnapshot(t, controller, func(snapshot SessionSnapshot) bool {
+		return snapshot.Desired == initial
+	})
+	if snapshot.Generation != 1 || snapshot.Plan.Topology != TopologyIndependent {
+		t.Fatalf("initial snapshot = %#v", snapshot)
+	}
+
+	newVideo := FeedConfig{Domain: "/new-video", UUID: "new-video", Active: true}
+	commands <- SessionCommand{Kind: CommandSetVideo, Config: newVideo}
+	receiveControllerEvent(t, events)
+	snapshot = waitControllerSnapshot(t, controller, func(snapshot SessionSnapshot) bool {
+		return snapshot.Desired.Video == newVideo
+	})
+	if snapshot.Generation != 1 || snapshot.Plan.Topology != TopologyIndependent {
+		t.Fatalf("same-topology snapshot = %#v", snapshot)
+	}
+
+	commands <- SessionCommand{Kind: CommandEnableSync}
+	for {
+		if event := receiveControllerEvent(t, events); event.unit == UnitSync && event.action == "start" {
+			break
+		}
+	}
+	snapshot = waitControllerSnapshot(t, controller, func(snapshot SessionSnapshot) bool {
+		return snapshot.Plan.Topology == TopologySynchronized
+	})
+	if snapshot.Generation != 2 || !snapshot.Desired.SyncRequested {
+		t.Fatalf("sync snapshot = %#v", snapshot)
+	}
+
+	commands <- SessionCommand{Kind: CommandDisableSync}
+	for {
+		event := receiveControllerEvent(t, events)
+		if event.action == "start" && (event.unit == UnitVideo || event.unit == UnitAudio) {
+			break
+		}
+	}
+	snapshot = waitControllerSnapshot(t, controller, func(snapshot SessionSnapshot) bool {
+		return snapshot.Plan.Topology == TopologyIndependent && !snapshot.Desired.SyncRequested
+	})
+	if snapshot.Generation != 3 {
+		t.Fatalf("independent snapshot generation = %d, want 3", snapshot.Generation)
+	}
+
+	close(commands)
+	if err := <-done; err != nil {
+		t.Fatalf("Run() error = %v", err)
+	}
+}
+
+func TestSessionControllerSnapshotRetainsUnavailableSyncRequest(t *testing.T) {
+	events := make(chan controllerEvent, 32)
+	controller, err := NewSessionController(
+		recordingVideoSlot{events},
+		recordingAudioSlot{events},
+		recordingSyncSlot{events},
+		nil,
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	initial := validCommandSession()
+	initial.SyncRequested = false
+	commands := make(chan SessionCommand)
+	done := make(chan error, 1)
+	go func() { done <- controller.Run(context.Background(), initial, commands) }()
+	receiveIndependentStarts(t, events)
+	commands <- SessionCommand{Kind: CommandEnableSync}
+
+	snapshot := waitControllerSnapshot(t, controller, func(snapshot SessionSnapshot) bool {
+		return snapshot.Desired.SyncRequested
+	})
+	if snapshot.Plan.Topology != TopologyIndependent || snapshot.Generation != 1 {
+		t.Fatalf("unsupported-sync snapshot = %#v", snapshot)
+	}
+	close(commands)
+	if err := <-done; err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestSessionControllerSnapshotConcurrentReads(t *testing.T) {
+	events := make(chan controllerEvent, 32)
+	controller := newRecordingController(t, events)
+	initial := validCommandSession()
+	initial.SyncRequested = false
+	commands := make(chan SessionCommand)
+	done := make(chan error, 1)
+	go func() { done <- controller.Run(context.Background(), initial, commands) }()
+	receiveIndependentStarts(t, events)
+	waitControllerSnapshot(t, controller, func(SessionSnapshot) bool { return true })
+
+	var readers sync.WaitGroup
+	for range 8 {
+		readers.Add(1)
+		go func() {
+			defer readers.Done()
+			for range 1_000 {
+				controller.Snapshot()
+			}
+		}()
+	}
+	readers.Wait()
+	close(commands)
+	if err := <-done; err != nil {
+		t.Fatal(err)
+	}
+}