diff --git a/internal/playback/session_controller.go b/internal/playback/session_controller.go index 94181b7..85879c1 100644 --- a/internal/playback/session_controller.go +++ b/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() +} diff --git a/internal/playback/session_controller_test.go b/internal/playback/session_controller_test.go index 38ad8c9..68dcdc4 100644 --- a/internal/playback/session_controller_test.go +++ b/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) + } +}