fixed the synchronized-audio corruption

cmd/mxl-player/main.go

@@ -6,16 +6,12 @@ import (
 	"fmt"
 	"io"
 	"log"
-	mxladapter "mxl-player/internal/adapter/mxl"
 	"mxl-player/internal/imgui"
-	"mxl-player/internal/output"
 	"mxl-player/internal/playback"
 	"mxl-player/internal/renderer"
 	"mxl-player/internal/sdl"
-	"mxl-player/internal/source"
 	"os"
 	"runtime"
-	"sync"
 	"time"
 	"unsafe"
 
@@ -170,22 +166,6 @@ func main() {
 	if !args.ListAudio && !args.ListGPU {
 		checkMXLargs(args)
 	}
-	if args.SyncRequested &&
-		args.VideoFlowId != "" &&
-		args.AudioFlowId != "" &&
-		args.VideoDomain != args.AudioDomain {
-		fmt.Fprintln(
-			os.Stderr,
-			"--sync currently requires video and audio to use the same MXL domain",
-		)
-		os.Exit(2)
-	}
-	// path selection
-	useLegacySync := args.SyncRequested &&
-		args.VideoFlowId != "" &&
-		args.AudioFlowId != ""
-	useIndependentSlots := !useLegacySync
-
 	runtime.LockOSThread()
 	if err := sdl.Load(); err != nil {
 		panic(err)
@@ -289,77 +269,6 @@ func main() {
 	}
 	defer vkDevice.Destroy()
 
-	var (
-		syncSrc     *source.SyncSource
-		videoSrc    *source.Source
-		audioSrc    *source.AudioSource
-		audioStream uintptr
-		audioBatch  uint64
-		aChans      uint64
-	)
-
-	interleaveAudio := func(samples [][]byte) []byte {
-		frameBytes := int(audioBatch) * int(aChans) * 4
-		out := make([]byte, frameBytes)
-		for ch := uint64(0); ch < aChans; ch++ {
-			srcBytes := samples[ch]
-			for i := uint64(0); i < audioBatch; i++ {
-				srcOff := i * 4
-				dstOff := (i*aChans + ch) * 4
-				if srcOff+4 <= uint64(len(srcBytes)) {
-					copy(out[dstOff:dstOff+4], srcBytes[srcOff:srcOff+4])
-				}
-			}
-		}
-		return out
-	}
-
-	switch {
-	case useLegacySync:
-		syncSrc, err = source.OpenSameDomainSync(
-			args.VideoDomain,
-			args.VideoFlowId,
-			args.AudioFlowId,
-		)
-		if err != nil {
-			log.Fatalf("sync source: %v", err)
-		}
-		aChans = syncSrc.Channels()
-		audioBatch = uint64(syncSrc.AudioRate().Num) / uint64(syncSrc.Rate().Num)
-		if audioBatch == 0 {
-			audioBatch = 1
-		}
-		audioStream = sdl.OpenAudioDeviceStream(sdlAudioDevice, sdl.AudioSpec{
-			Format:   sdl.AudioF32,
-			Channels: int32(aChans),
-			Freq:     int32(syncSrc.AudioRate().Num / syncSrc.AudioRate().Den),
-		})
-		if audioStream == 0 {
-			log.Fatalf("audio: %s", sdl.GetError())
-		}
-		sdl.ResumeAudioStreamDevice(audioStream)
-		fmt.Printf("sync: video %dx%d audio %dch batch=%d\n",
-			syncSrc.Width(), syncSrc.Height(), aChans, audioBatch)
-	default:
-		// Independent slots own their readers.
-		// Empty startup opens nothing.
-	}
-
-	defer func() {
-		if syncSrc != nil {
-			_ = syncSrc.Close()
-		}
-		if videoSrc != nil {
-			_ = videoSrc.Close()
-		}
-		if audioSrc != nil {
-			_ = audioSrc.Close()
-		}
-	}()
-	if audioStream != 0 {
-		defer sdl.DestroyAudioStream(audioStream)
-	}
-
 	// Create renderer
 	r, err := renderer.New(renderer.Config{
 		PhysDevice:     vkPhysDevice,
@@ -414,419 +323,58 @@ func main() {
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	type reconnectParams struct {
+	player, err := newPlayerPlayback(sdlAudioDevice, retryPolicy)
-		domain string
-		video  string
-		audio  string
-	}
-	// Reconnect requests from GUI or automatic retry
-	control := make(chan reconnectParams, 1)
-	videoBridge := playback.NewVideoBridge()
-	statusStore := playback.NewStatusStore()
-
-	videoWorker, err := playback.NewVideoWorker(
-		mxladapter.VideoFactory{},
-		videoBridge,
-		retryPolicy,
-		mxladapter.ShouldRetry,
-		func(status playback.Status) {
-			statusStore.Observe(status)
-			if status.Err != nil {
-				log.Printf(
-					"video: state=%v attempt=%d failed=%d: %v",
-					status.State,
-					status.Attempt,
-					status.FailedAttempts,
-					status.Err,
-				)
-				return
-			}
-			log.Printf(
-				"video: state=%v attempt=%d failed=%d",
-				status.State,
-				status.Attempt,
-				status.FailedAttempts,
-			)
-		},
-	)
 	if err != nil {
 		panic(err)
 	}
-	videoSlot, err := playback.NewVideoSlot(videoWorker)
+	videoBridge := player.Video
-	if err != nil {
+	statusStore := player.Status
-		panic(err)
+	syncRequested := args.SyncRequested
-	}
-	videoCommands := make(chan playback.FeedConfig, 1)
 
-	audioOutput := output.NewSDLAudioSink(sdlAudioDevice)
+	enqueueCommand := func(command playback.SessionCommand) {
-	defer audioOutput.Close()
-	audioWorker, err := playback.NewAudioWorker(
-		mxladapter.AudioFactory{},
-		audioOutput,
-		retryPolicy,
-		mxladapter.ShouldRetry,
-		func(status playback.Status) {
-			statusStore.Observe(status)
-
-			if status.Err != nil {
-				log.Printf(
-					"audio: state=%v attempt=%d failed=%d: %v",
-					status.State,
-					status.Attempt,
-					status.FailedAttempts,
-					status.Err,
-				)
-				return
-			}
-
-			log.Printf(
-				"audio: state=%v attempt=%d failed=%d",
-				status.State,
-				status.Attempt,
-				status.FailedAttempts,
-			)
-		},
-	)
-	if err != nil {
-		panic(err)
-	}
-	audioSlot, err := playback.NewAudioSlot(audioWorker)
-	if err != nil {
-		panic(err)
-	}
-	audioCommands := make(chan playback.FeedConfig, 1)
-
-	reopen := func(params reconnectParams) error {
-		// Close current sources
-		if syncSrc != nil {
-			_ = syncSrc.Close()
-			syncSrc = nil
-		}
-		if videoSrc != nil {
-			_ = videoSrc.Close()
-			videoSrc = nil
-		}
-		if audioSrc != nil {
-			_ = audioSrc.Close()
-			audioSrc = nil
-		}
-		// Try once. Return error if fails — caller loops back to select
-		// and can pick up newer reconnect request.
-		if params.video != "" && params.audio != "" {
-			s, e := source.OpenSameDomainSync(params.domain, params.video, params.audio)
-			if e == nil {
-				syncSrc = s
-				aChans = s.Channels()
-				audioBatch = uint64(s.AudioRate().Num) / uint64(s.Rate().Num)
-				if audioBatch == 0 {
-					audioBatch = 1
-				}
-				return nil
-			}
-			return e
-		} else if params.video != "" {
-			s, e := source.Open(params.domain, params.video)
-			if e == nil {
-				videoSrc = s
-				return nil
-			}
-			return e
-		} else if params.audio != "" {
-			s, e := source.OpenAudio(params.domain, params.audio)
-			if e == nil {
-				audioSrc = s
-				aChans = s.Channels()
-				audioBatch = uint64(s.Rate().Num) / (100 * uint64(s.Rate().Den))
-				if audioBatch == 0 {
-					audioBatch = 1
-				}
-				return nil
-			}
-			return e
-		}
-		return fmt.Errorf("reopen: no flow specified")
-	}
-
-	enqueueVideoConfig := func(config playback.FeedConfig) {
 		select {
-		case <-videoCommands:
+		case player.Commands <- command:
-		default:
-		}
-
-		select {
-		case videoCommands <- config:
 		default:
+			log.Printf("playback command queue is full; ignoring command %d", command.Kind)
 		}
 	}
-	enqueueAudioConfig := func(config playback.FeedConfig) {
-		select {
-		case <-audioCommands:
-		default:
-		}
-
-		select {
-		case audioCommands <- config:
-		default:
-		}
-	}
-
 	doReconnect := func() {
-		if useIndependentSlots {
 		videoActive = videoStr != ""
 		audioActive = audioStr != ""
 
-			videoConfig := playback.FeedConfig{}
+		if videoStr == "" {
-			if videoStr != "" {
+			enqueueCommand(playback.SessionCommand{Kind: playback.CommandRemoveVideo})
-				videoConfig = playback.FeedConfig{
+		} else {
+			enqueueCommand(playback.SessionCommand{
+				Kind: playback.CommandSetVideo,
+				Config: playback.FeedConfig{
 					Domain: videoDomainStr,
 					UUID:   videoStr,
 					Active: true,
+				},
+			})
 		}
-			}
+		if audioStr == "" {
-
+			enqueueCommand(playback.SessionCommand{Kind: playback.CommandRemoveAudio})
-			audioConfig := playback.FeedConfig{}
+		} else {
-			if audioStr != "" {
+			enqueueCommand(playback.SessionCommand{
-				audioConfig = playback.FeedConfig{
+				Kind: playback.CommandSetAudio,
+				Config: playback.FeedConfig{
 					Domain: audioDomainStr,
 					UUID:   audioStr,
 					Active: true,
-				}
-			}
-
-			enqueueVideoConfig(videoConfig)
-			enqueueAudioConfig(audioConfig)
-			return
-		}
-		// legacy
-		if videoDomainStr != audioDomainStr {
-			log.Printf(
-				"sync reconnect rejected: video domain %q differs from audio domain %q",
-				videoDomainStr,
-				audioDomainStr,
-			)
-			return
-		}
-		select {
-		case <-control:
-		default:
-		}
-		control <- reconnectParams{domain: videoDomainStr, video: videoStr, audio: audioStr}
-	}
-
-	playbackDone := make(chan struct{})
-	go func() {
-		defer close(playbackDone)
-
-		if useIndependentSlots {
-			var slots sync.WaitGroup
-			slots.Add(2)
-
-			go func() {
-				defer slots.Done()
-
-				err := videoSlot.Run(
-					ctx,
-					playback.FeedConfig{
-						Domain: args.VideoDomain,
-						UUID:   args.VideoFlowId,
-						Active: args.VideoFlowId != "",
 				},
-					videoCommands,
+			})
-				)
+		}
-				if err != nil && !errors.Is(err, context.Canceled) {
-					log.Printf("video slot: %v", err)
 	}
-			}()
 
+	playbackDone := make(chan error, 1)
 	go func() {
-				defer slots.Done()
+		playbackDone <- player.Controller.Run(
-
-				err := audioSlot.Run(
 			ctx,
-					playback.FeedConfig{
+			args.playbackConfig(),
-						Domain: args.AudioDomain,
+			player.Commands,
-						UUID:   args.AudioFlowId,
-						Active: args.AudioFlowId != "",
-					},
-					audioCommands,
 		)
-				if err != nil && !errors.Is(err, context.Canceled) {
-					log.Printf("audio slot: %v", err)
-				}
-			}()
-
-			<-ctx.Done()
-			slots.Wait()
-			return
-		}
-
-		// Audio-only mode: independent loop.
-		if audioSrc != nil && syncSrc == nil && videoSrc == nil {
-			for {
-				select {
-				case <-ctx.Done():
-					return
-				case params := <-control:
-					if params.video != "" || params.audio != "" {
-						if rerr := reopen(params); rerr != nil {
-							if errors.Is(rerr, context.Canceled) {
-								return
-							}
-							log.Printf("source: reopen failed: %v, retrying", rerr)
-							select {
-							case <-time.After(500 * time.Millisecond):
-							case <-ctx.Done():
-								return
-							}
-							select {
-							case control <- params:
-							default:
-							}
-						}
-					}
-					continue
-				default:
-				}
-				queued := sdl.GetAudioStreamQueued(audioStream)
-				maxQueued := int32(audioBatch) * int32(aChans) * 4 * 20
-				if queued > maxQueued {
-					select {
-					case <-time.After(10 * time.Millisecond):
-					case <-ctx.Done():
-						return
-					}
-					continue
-				}
-				f, err := audioSrc.NextAudio(ctx, audioBatch, 20*time.Millisecond)
-				if err != nil {
-					if errors.Is(err, context.Canceled) {
-						return
-					}
-					log.Printf("source: %v", err)
-					params := reconnectParams{domain: videoDomainStr, video: videoStr, audio: audioStr}
-					select {
-					case <-control:
-					default:
-					}
-					select {
-					case <-time.After(500 * time.Millisecond):
-					case <-ctx.Done():
-						return
-					}
-					select {
-					case control <- params:
-					default:
-					}
-					continue
-				}
-				if f.Samples != nil && audioStream != 0 {
-					sdl.PutAudioStreamData(audioStream, interleaveAudio(f.Samples))
-				}
-			}
-		}
-
-		// Video bridge provides backpressure: only one borrowed frame is in flight.
-		for {
-			select {
-			case <-ctx.Done():
-				return
-
-			case params := <-control:
-				if rerr := reopen(params); rerr != nil {
-					if errors.Is(rerr, context.Canceled) {
-						return
-					}
-					log.Printf("source: reopen failed: %v, retrying", rerr)
-
-					select {
-					case <-time.After(500 * time.Millisecond):
-					case <-ctx.Done():
-						return
-					}
-
-					select {
-					case control <- params:
-					default:
-						// Preserve an already queued, potentially newer request.
-					}
-				}
-				continue
-
-			default:
-			}
-
-			var videoFrame playback.VideoFrame
-			if syncSrc != nil {
-				vFrame, aFrame, err := syncSrc.NextSync(ctx, audioBatch, 200*time.Millisecond)
-				if err != nil {
-					if errors.Is(err, context.Canceled) {
-						return
-					}
-					log.Printf("source: %v", err)
-					// Request a reconnect after the read failure.
-					params := reconnectParams{
-						domain: videoDomainStr,
-						video:  videoStr,
-						audio:  audioStr,
-					}
-					select {
-					case control <- params:
-					default:
-						// Preserve an already queued, potentially newer request.
-					}
-					continue
-				}
-				videoFrame = playback.VideoFrame{
-					Index:   vFrame.Index,
-					Width:   vFrame.Width,
-					Height:  vFrame.Height,
-					Stride:  vFrame.Stride,
-					Size:    vFrame.Size,
-					Invalid: vFrame.Invalid,
-					Payload: vFrame.Payload,
-				}
-				if aFrame.Samples != nil && audioStream != 0 {
-					sdl.PutAudioStreamData(audioStream, interleaveAudio(aFrame.Samples))
-				}
-
-			} else if videoSrc != nil {
-				f, err := videoSrc.NextCtx(ctx, 200*time.Millisecond)
-				if err != nil {
-					if errors.Is(err, context.Canceled) {
-						return
-					}
-					log.Printf("source: %v", err)
-					params := reconnectParams{
-						domain: videoDomainStr,
-						video:  videoStr,
-						audio:  audioStr,
-					}
-					select {
-					case control <- params:
-					default:
-						// Preserve an already queued, potentially newer request.
-					}
-					continue
-				}
-				videoFrame = playback.VideoFrame{
-					Index:   f.Index,
-					Width:   f.Width,
-					Height:  f.Height,
-					Stride:  f.Stride,
-					Size:    f.Size,
-					Invalid: f.Invalid,
-					Payload: f.Payload,
-				}
-			}
-
-			if err := videoBridge.ConsumeVideo(ctx, videoFrame); err != nil {
-				if errors.Is(err, context.Canceled) {
-					return
-				}
-				log.Printf("video output: %v", err)
-				return
-			}
-		}
 	}()
 
 	running := true
@@ -979,41 +527,43 @@ func main() {
 			cimgui.InputTextWithHint("Audio domain", "/dev/shm/mxl", &audioDomainStr, 0, nil)
 			cimgui.InputTextWithHint("Video UUID", "", &videoStr, 0, nil)
 			cimgui.InputTextWithHint("Audio UUID", "", &audioStr, 0, nil)
+			if snapshot, ok := player.Controller.Snapshot(); ok {
+				videoActive = snapshot.Desired.Video.Active
+				audioActive = snapshot.Desired.Audio.Active
+				syncRequested = snapshot.Desired.SyncRequested
+			}
 			if cimgui.Button("Connect") {
 				doReconnect()
 			}
 			cimgui.SameLine()
 			cimgui.Checkbox("Show stats", &showStats)
-			if useIndependentSlots && videoActive {
+			if cimgui.Checkbox("Synchronize", &syncRequested) {
+				kind := playback.CommandDisableSync
+				if syncRequested {
+					kind = playback.CommandEnableSync
+				}
+				enqueueCommand(playback.SessionCommand{Kind: kind})
+			}
+			if videoActive {
 				if cimgui.Button("Stop video") {
 					videoActive = false
-					enqueueVideoConfig(
+					enqueueCommand(playback.SessionCommand{Kind: playback.CommandStopVideo})
-						playback.FeedConfig{
-							Domain: videoDomainStr,
-							UUID:   videoStr,
-							Active: false,
-						})
 				}
 			}
-			if useIndependentSlots && !videoActive && videoStr != "" {
+			if !videoActive && videoStr != "" {
 				cimgui.SameLine()
 				if cimgui.Button("Resume video") {
 					videoActive = true
-					enqueueVideoConfig(playback.FeedConfig{
+					enqueueCommand(playback.SessionCommand{Kind: playback.CommandResumeVideo})
-						Domain: videoDomainStr,
-						UUID:   videoStr,
-						Active: true,
-					})
 				}
 			}
-			if useIndependentSlots && videoStr != "" {
+			if videoStr != "" {
 				if cimgui.Button("Remove video") {
 					videoActive = false
 					videoStr = ""
-					enqueueVideoConfig(playback.FeedConfig{})
+					enqueueCommand(playback.SessionCommand{Kind: playback.CommandRemoveVideo})
 				}
 			}
-			if useIndependentSlots {
 			if videoActive {
 				cimgui.Text("Video desired: active")
 			} else if videoStr != "" {
@@ -1023,61 +573,39 @@ func main() {
 			}
 
 			if status, ok := statusStore.Snapshot(playback.UnitVideo); ok {
-					cimgui.Text(fmt.Sprintf(
+				cimgui.Text(fmt.Sprintf("Video actual: %s", status.State))
-						"Video actual: %s",
+				cimgui.Text(fmt.Sprintf("Attempt: %d, failed: %d", status.Attempt, status.FailedAttempts))
-						status.State,
-					))
-					cimgui.Text(fmt.Sprintf(
-						"Attempt: %d, failed: %d",
-						status.Attempt,
-						status.FailedAttempts,
-					))
-
 				if status.RetryIn > 0 {
-						cimgui.Text(fmt.Sprintf(
+					cimgui.Text(fmt.Sprintf("Retry in: %s", status.RetryIn.Round(time.Millisecond)))
-							"Retry in: %s",
-							status.RetryIn.Round(time.Millisecond),
-						))
 				}
-
 				if status.Err != nil {
 					cimgui.TextWrapped(status.Err.Error())
 				}
 			} else {
 				cimgui.Text("Video actual: not started")
 			}
-			}
+			if audioActive {
-			if useIndependentSlots && audioActive {
 				if cimgui.Button("Stop audio") {
 					audioActive = false
-					enqueueAudioConfig(playback.FeedConfig{
+					enqueueCommand(playback.SessionCommand{Kind: playback.CommandStopAudio})
-						Domain: audioDomainStr,
-						UUID:   audioStr,
-						Active: false,
-					})
 				}
 			}
 
-			if useIndependentSlots && !audioActive && audioStr != "" {
+			if !audioActive && audioStr != "" {
 				if cimgui.Button("Resume audio") {
 					audioActive = true
-					enqueueAudioConfig(playback.FeedConfig{
+					enqueueCommand(playback.SessionCommand{Kind: playback.CommandResumeAudio})
-						Domain: audioDomainStr,
-						UUID:   audioStr,
-						Active: true,
-					})
 				}
 			}
 
-			if useIndependentSlots && audioStr != "" {
+			if audioStr != "" {
 				if cimgui.Button("Remove audio") {
 					audioActive = false
 					audioStr = ""
-					enqueueAudioConfig(playback.FeedConfig{})
+					enqueueCommand(playback.SessionCommand{Kind: playback.CommandRemoveAudio})
 				}
 			}
 
-			if useIndependentSlots {
 			if audioActive {
 				cimgui.Text("Audio desired: active")
 			} else if audioStr != "" {
@@ -1087,29 +615,22 @@ func main() {
 			}
 
 			if status, ok := statusStore.Snapshot(playback.UnitAudio); ok {
-					cimgui.Text(fmt.Sprintf(
+				cimgui.Text(fmt.Sprintf("Audio actual: %s", status.State))
-						"Audio actual: %s",
+				cimgui.Text(fmt.Sprintf("Attempt: %d, failed: %d", status.Attempt, status.FailedAttempts))
-						status.State,
-					))
-					cimgui.Text(fmt.Sprintf(
-						"Attempt: %d, failed: %d",
-						status.Attempt,
-						status.FailedAttempts,
-					))
-
 				if status.RetryIn > 0 {
-						cimgui.Text(fmt.Sprintf(
+					cimgui.Text(fmt.Sprintf("Retry in: %s", status.RetryIn.Round(time.Millisecond)))
-							"Retry in: %s",
-							status.RetryIn.Round(time.Millisecond),
-						))
 				}
-
 				if status.Err != nil {
 					cimgui.TextWrapped(status.Err.Error())
 				}
 			} else {
 				cimgui.Text("Audio actual: not started")
 			}
+			if status, ok := statusStore.Snapshot(playback.UnitSync); ok {
+				cimgui.Text(fmt.Sprintf("Sync actual: %s", status.State))
+				if status.Err != nil {
+					cimgui.TextWrapped(status.Err.Error())
+				}
 			}
 
 			cimgui.End()
@@ -1141,5 +662,10 @@ func main() {
 	}
 
 	cancel()
-	<-playbackDone
+	if err := <-playbackDone; err != nil && !errors.Is(err, context.Canceled) {
+		log.Printf("playback controller: %v", err)
+	}
+	if err := player.Close(); err != nil {
+		log.Printf("close playback: %v", err)
+	}
 }

cmd/mxl-player/playback_runtime.go

@@ -0,0 +1,127 @@
+package main
+
+import (
+	"log"
+
+	mxladapter "mxl-player/internal/adapter/mxl"
+	"mxl-player/internal/output"
+	"mxl-player/internal/playback"
+)
+
+type playerPlayback struct {
+	Controller *playback.SessionController
+	Commands   chan playback.SessionCommand
+	Video      *playback.VideoBridge
+	Status     *playback.StatusStore
+	Audio      *output.SDLAudioSink
+}
+
+func newPlayerPlayback(
+	audioDevice uint32,
+	retry playback.RetryPolicy,
+) (*playerPlayback, error) {
+	videoBridge := playback.NewVideoBridge()
+	statusStore := playback.NewStatusStore()
+	audioSink := output.NewSDLAudioSink(audioDevice)
+
+	observe := func(status playback.Status) {
+		statusStore.Observe(status)
+
+		if status.Err != nil {
+			log.Printf(
+				"%s: state=%v attempt=%d failed=%d: %v",
+				status.Unit,
+				status.State,
+				status.Attempt,
+				status.FailedAttempts,
+				status.Err,
+			)
+			return
+		}
+
+		log.Printf(
+			"%s: state=%v attempt=%d failed=%d",
+			status.Unit,
+			status.State,
+			status.Attempt,
+			status.FailedAttempts,
+		)
+	}
+
+	videoWorker, err := playback.NewVideoWorker(
+		mxladapter.VideoFactory{},
+		videoBridge,
+		retry,
+		mxladapter.ShouldRetry,
+		observe,
+	)
+	if err != nil {
+		_ = audioSink.Close()
+		return nil, err
+	}
+	videoSlot, err := playback.NewVideoSlot(videoWorker)
+	if err != nil {
+		_ = audioSink.Close()
+		return nil, err
+	}
+
+	audioWorker, err := playback.NewAudioWorker(
+		mxladapter.AudioFactory{},
+		audioSink,
+		retry,
+		mxladapter.ShouldRetry,
+		observe,
+	)
+	if err != nil {
+		_ = audioSink.Close()
+		return nil, err
+	}
+	audioSlot, err := playback.NewAudioSlot(audioWorker)
+	if err != nil {
+		_ = audioSink.Close()
+		return nil, err
+	}
+
+	syncWorker, err := playback.NewSyncWorker(
+		mxladapter.SyncFactory{},
+		videoBridge,
+		audioSink,
+		retry,
+		mxladapter.ShouldRetry,
+		observe,
+	)
+	if err != nil {
+		_ = audioSink.Close()
+		return nil, err
+	}
+	syncSlot, err := playback.NewSyncSlot(syncWorker)
+	if err != nil {
+		_ = audioSink.Close()
+		return nil, err
+	}
+
+	controller, err := playback.NewSessionController(
+		videoSlot,
+		audioSlot,
+		syncSlot,
+		func(video, audio playback.FeedConfig) bool {
+			return video.Domain == audio.Domain
+		},
+	)
+	if err != nil {
+		_ = audioSink.Close()
+		return nil, err
+	}
+
+	return &playerPlayback{
+		Controller: controller,
+		Commands:   make(chan playback.SessionCommand, 32),
+		Video:      videoBridge,
+		Status:     statusStore,
+		Audio:      audioSink,
+	}, nil
+}
+
+func (p *playerPlayback) Close() error {
+	return p.Audio.Close()
+}

cmd/mxl-player/playback_runtime_test.go

@@ -0,0 +1,42 @@
+package main
+
+import (
+	"testing"
+	"time"
+
+	"mxl-player/internal/playback"
+)
+
+func validRuntimeRetryPolicy() playback.RetryPolicy {
+	return playback.RetryPolicy{
+		MaxAttempts:  3,
+		InitialDelay: time.Millisecond,
+		MaxDelay:     time.Second,
+	}
+}
+
+func TestNewPlayerPlaybackBuildsCompleteRuntime(t *testing.T) {
+	runtime, err := newPlayerPlayback(123, validRuntimeRetryPolicy())
+	if err != nil {
+		t.Fatalf("newPlayerPlayback() error = %v", err)
+	}
+	if runtime.Controller == nil || runtime.Video == nil || runtime.Status == nil || runtime.Audio == nil {
+		t.Fatalf("newPlayerPlayback() = %#v", runtime)
+	}
+	if runtime.Commands == nil || cap(runtime.Commands) != 32 {
+		t.Fatalf("command channel = %#v, capacity = %d", runtime.Commands, cap(runtime.Commands))
+	}
+	if err := runtime.Close(); err != nil {
+		t.Fatalf("Close() error = %v", err)
+	}
+}
+
+func TestNewPlayerPlaybackRejectsInvalidRetryPolicy(t *testing.T) {
+	runtime, err := newPlayerPlayback(123, playback.RetryPolicy{})
+	if runtime != nil {
+		t.Fatalf("newPlayerPlayback() runtime = %#v, want nil", runtime)
+	}
+	if err == nil {
+		t.Fatal("newPlayerPlayback() error is nil")
+	}
+}

imgui.ini

@@ -14,7 +14,7 @@ Size=200,200
 Collapsed=0
 
 [Window][Connection]
-Pos=475,529
+Pos=322,387
-Size=640,352
+Size=618,275
 Collapsed=0
 

internal/adapter/mxl/sync.go

@@ -14,7 +14,6 @@ import (
 
 const (
 	DefaultSyncReadTimeout = 200 * time.Millisecond
-	DefaultSyncBatchDuration = 10 * time.Millisecond
 )
 
 var ErrNativeSyncDifferentDomains = errors.New(
@@ -23,14 +22,12 @@ var ErrNativeSyncDifferentDomains = errors.New(
 
 type SyncFactory struct {
 	ReadTimeout time.Duration
-	BatchDuration time.Duration
 	open        func(string, string, string) (localSyncSource, error)
 }
 
 type localSyncSource interface {
 	NextSync(
 		context.Context,
-		uint64,
 		time.Duration,
 	) (source.Frame, source.AudioFrame, error)
 
@@ -41,7 +38,6 @@ type localSyncSource interface {
 type syncReader struct {
 	source          localSyncSource
 	readTimeout     time.Duration
-	audioBatch      uint64
 	rateNumerator   int64
 	rateDenominator int64
 }
@@ -118,25 +114,11 @@ func (f SyncFactory) OpenSync(
 	if readTimeout <= 0 {
 		readTimeout = DefaultSyncReadTimeout
 	}
-	batchDuration := f.BatchDuration
-	if batchDuration <= 0 {
-		batchDuration = DefaultSyncBatchDuration
-	}
 	audioRate := src.AudioRate()
-	batch, err := audioBatchSize(audioRate.Num, audioRate.Den, batchDuration)
-	if err != nil {
-		_ = src.Close()
-		return nil, &source.SourceError{
-			Op:   "calculate sync audio batch",
-			Kind: source.ErrorKindInvalidConfig,
-			Err:  err,
-		}
-	}
 
 	return &syncReader{
 		source:          src,
 		readTimeout:     readTimeout,
-		audioBatch:      batch,
 		rateNumerator:   audioRate.Num,
 		rateDenominator: audioRate.Den,
 	}, nil
@@ -145,7 +127,7 @@ func (f SyncFactory) OpenSync(
 func (r *syncReader) ReadSync(
 	ctx context.Context,
 ) (playback.SyncFrame, error) {
-	video, audio, err := r.source.NextSync(ctx, r.audioBatch, r.readTimeout)
+	video, audio, err := r.source.NextSync(ctx, r.readTimeout)
 	if err != nil {
 		return playback.SyncFrame{}, err
 	}

internal/adapter/mxl/sync_test.go

@@ -17,7 +17,6 @@ type fakeLocalSyncSource struct {
 	audio      source.AudioFrame
 	readErr    error
 	rate       mxl.Rational
-	batch      uint64
 	timeout    time.Duration
 	closed     bool
 	closeError error
@@ -25,10 +24,8 @@ type fakeLocalSyncSource struct {
 
 func (s *fakeLocalSyncSource) NextSync(
 	_ context.Context,
-	batch uint64,
 	timeout time.Duration,
 ) (source.Frame, source.AudioFrame, error) {
-	s.batch = batch
 	s.timeout = timeout
 	return s.video, s.audio, s.readErr
 }
@@ -79,23 +76,8 @@ func TestSyncFactoryUsesDefaultsAndForwardsFeeds(t *testing.T) {
 		t.Fatalf("open args = %q %q %q", domain, videoUUID, audioUUID)
 	}
 	got := reader.(*syncReader)
-	if got.readTimeout != DefaultSyncReadTimeout || got.audioBatch != 480 {
+	if got.readTimeout != DefaultSyncReadTimeout {
-		t.Fatalf("reader timeout=%s batch=%d, want %s and 480", got.readTimeout, got.audioBatch, DefaultSyncReadTimeout)
+		t.Fatalf("reader timeout=%s, want %s", got.readTimeout, DefaultSyncReadTimeout)
-	}
-}
-
-func TestSyncFactoryClosesSourceForInvalidAudioRate(t *testing.T) {
-	fake := &fakeLocalSyncSource{rate: mxl.Rational{}}
-	factory := SyncFactory{open: func(string, string, string) (localSyncSource, error) {
-		return fake, nil
-	}}
-	video, audio := syncFeedConfigs()
-	reader, err := factory.OpenSync(context.Background(), video, audio)
-	if reader != nil {
-		t.Fatal("OpenSync() reader is not nil")
-	}
-	if !errors.Is(err, ErrInvalidAudioBatch) || !fake.closed {
-		t.Fatalf("OpenSync() error=%v closed=%t", err, fake.closed)
 	}
 }
 
@@ -108,7 +90,7 @@ func TestSyncReaderConvertsPairWithoutCopying(t *testing.T) {
 		rate:  mxl.Rational{Num: 48_000, Den: 1},
 	}
 	reader := &syncReader{
-		source: fake, readTimeout: 7 * time.Millisecond, audioBatch: 12,
+		source: fake, readTimeout: 7 * time.Millisecond,
 		rateNumerator: 48_000, rateDenominator: 1,
 	}
 
@@ -116,8 +98,8 @@ func TestSyncReaderConvertsPairWithoutCopying(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if fake.batch != 12 || fake.timeout != 7*time.Millisecond {
+	if fake.timeout != 7*time.Millisecond {
-		t.Fatalf("NextSync() batch=%d timeout=%s", fake.batch, fake.timeout)
+		t.Fatalf("NextSync() timeout=%s", fake.timeout)
 	}
 	if frame.Video.Index != 10 || frame.Audio.Index != 40 || frame.Audio.SampleRateNumerator != 48_000 {
 		t.Fatalf("frame = %+v", frame)

internal/playback/session_controller.go

@@ -0,0 +1,344 @@
+package playback
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+)
+
+type VideoSlotRunner interface {
+	Run(
+		context.Context,
+		FeedConfig,
+		<-chan FeedConfig,
+	) error
+}
+
+type AudioSlotRunner interface {
+	Run(
+		context.Context,
+		FeedConfig,
+		<-chan FeedConfig,
+	) error
+}
+
+type SyncSlotRunner interface {
+	Run(
+		context.Context,
+		SyncPairConfig,
+		<-chan SyncPairConfig,
+	) error
+}
+
+var _ VideoSlotRunner = (*VideoSlot)(nil)
+var _ AudioSlotRunner = (*AudioSlot)(nil)
+var _ SyncSlotRunner = (*SyncSlot)(nil)
+
+var (
+	ErrVideoSlotRequired = errors.New("video slot is required")
+	ErrAudioSlotRequired = errors.New("audio slot is required")
+	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(
+	videoSlot VideoSlotRunner,
+	audioSlot AudioSlotRunner,
+	syncSlot SyncSlotRunner,
+	canSync SyncPredicate,
+) (*SessionController, error) {
+	if videoSlot == nil {
+		return nil, ErrVideoSlotRequired
+	}
+	if audioSlot == nil {
+		return nil, ErrAudioSlotRequired
+	}
+	if syncSlot == nil {
+		return nil, ErrSyncSlotRequired
+	}
+	return &SessionController{
+		videoSlot: videoSlot,
+		audioSlot: audioSlot,
+		syncSlot:  syncSlot,
+		canSync:   canSync,
+	}, nil
+}
+
+type sessionRuntime struct {
+	topology SessionTopology
+	cancel   context.CancelFunc
+	done     chan struct{}
+	result   error
+
+	videoCommands chan FeedConfig
+	audioCommands chan FeedConfig
+	syncCommands  chan SyncPairConfig
+}
+
+var (
+	ErrSessionRuntimeStopped = errors.New(
+		"session slot runtime stopped unexpectedly",
+	)
+)
+
+func (c *SessionController) Run(
+	ctx context.Context,
+	initial SessionConfig,
+	commands <-chan SessionCommand,
+) error {
+	plan, err := BuildSessionPlan(initial, c.canSync)
+	if err != nil {
+		return fmt.Errorf("build initial session plan: %w", err)
+	}
+
+	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 {
+			runtimeDone = runtime.done
+		}
+
+		select {
+		case <-ctx.Done():
+			stopSessionRuntime(runtime)
+			return ctx.Err()
+
+		case <-runtimeDone:
+			return unexpectedSessionRuntimeError(runtime.result)
+
+		case command, ok := <-commands:
+			if !ok {
+				stopSessionRuntime(runtime)
+				return nil
+			}
+
+			nextDesired, err := ApplySessionCommand(desired, command)
+			if err != nil {
+				// Invalid commands must not disturb the current valid runtime.
+				continue
+			}
+			nextPlan, err := BuildSessionPlan(nextDesired, c.canSync)
+			if err != nil {
+				continue
+			}
+
+			nextRuntime, err := c.reconcileSessionRuntime(
+				ctx,
+				runtime,
+				plan,
+				nextPlan,
+			)
+			if err != nil {
+				stopSessionRuntime(runtime)
+				if ctx.Err() != nil {
+					return ctx.Err()
+				}
+				return err
+			}
+
+			if plan.Topology != nextPlan.Topology {
+				generation++
+			}
+			desired = nextDesired
+			plan = nextPlan
+			runtime = nextRuntime
+			c.publish(SessionSnapshot{
+				Desired:    desired,
+				Plan:       plan,
+				Generation: generation,
+			})
+		}
+	}
+}
+
+func (c *SessionController) startSessionRuntime(
+	ctx context.Context,
+	plan SessionPlan,
+) *sessionRuntime {
+	if plan.Topology == TopologyIdle {
+		return &sessionRuntime{topology: TopologyIdle}
+	}
+
+	runtimeCtx, cancel := context.WithCancel(ctx)
+	runtime := &sessionRuntime{
+		topology: plan.Topology,
+		cancel:   cancel,
+		done:     make(chan struct{}),
+	}
+
+	switch plan.Topology {
+	case TopologyIndependent:
+		runtime.videoCommands = make(chan FeedConfig)
+		runtime.audioCommands = make(chan FeedConfig)
+		results := make(chan error, 2)
+
+		go func() {
+			results <- c.videoSlot.Run(
+				runtimeCtx,
+				plan.Video,
+				runtime.videoCommands,
+			)
+		}()
+		go func() {
+			results <- c.audioSlot.Run(
+				runtimeCtx,
+				plan.Audio,
+				runtime.audioCommands,
+			)
+		}()
+		go func() {
+			first := <-results
+			cancel()
+			second := <-results
+			runtime.result = errors.Join(first, second)
+			close(runtime.done)
+		}()
+
+	case TopologySynchronized:
+		runtime.syncCommands = make(chan SyncPairConfig)
+		go func() {
+			runtime.result = c.syncSlot.Run(
+				runtimeCtx,
+				plan.Sync,
+				runtime.syncCommands,
+			)
+			close(runtime.done)
+		}()
+	}
+
+	return runtime
+}
+
+func stopSessionRuntime(runtime *sessionRuntime) {
+	if runtime == nil || runtime.done == nil {
+		return
+	}
+	runtime.cancel()
+	<-runtime.done
+}
+
+func unexpectedSessionRuntimeError(err error) error {
+	if err == nil {
+		return ErrSessionRuntimeStopped
+	}
+	return fmt.Errorf("%w: %v", ErrSessionRuntimeStopped, err)
+}
+
+func (c *SessionController) reconcileSessionRuntime(
+	ctx context.Context,
+	runtime *sessionRuntime,
+	current SessionPlan,
+	next SessionPlan,
+) (*sessionRuntime, error) {
+	if current.Topology != next.Topology {
+		stopSessionRuntime(runtime)
+		if err := ctx.Err(); err != nil {
+			return runtime, err
+		}
+		return c.startSessionRuntime(ctx, next), nil
+	}
+
+	switch next.Topology {
+	case TopologyIndependent:
+		if current.Video != next.Video {
+			if !sendFeedConfig(ctx, runtime.done, runtime.videoCommands, next.Video) {
+				return runtime, sessionRuntimeSendError(ctx, runtime)
+			}
+		}
+		if current.Audio != next.Audio {
+			if !sendFeedConfig(ctx, runtime.done, runtime.audioCommands, next.Audio) {
+				return runtime, sessionRuntimeSendError(ctx, runtime)
+			}
+		}
+
+	case TopologySynchronized:
+		if current.Sync != next.Sync {
+			if !sendSyncPairConfig(ctx, runtime.done, runtime.syncCommands, next.Sync) {
+				return runtime, sessionRuntimeSendError(ctx, runtime)
+			}
+		}
+	}
+
+	return runtime, nil
+}
+
+func sendFeedConfig(
+	ctx context.Context,
+	done <-chan struct{},
+	commands chan<- FeedConfig,
+	config FeedConfig,
+) bool {
+	select {
+	case commands <- config:
+		return true
+	case <-ctx.Done():
+		return false
+	case <-done:
+		return false
+	}
+}
+
+func sendSyncPairConfig(
+	ctx context.Context,
+	done <-chan struct{},
+	commands chan<- SyncPairConfig,
+	config SyncPairConfig,
+) bool {
+	select {
+	case commands <- config:
+		return true
+	case <-ctx.Done():
+		return false
+	case <-done:
+		return false
+	}
+}
+
+func sessionRuntimeSendError(ctx context.Context, runtime *sessionRuntime) error {
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	<-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

@@ -0,0 +1,422 @@
+package playback
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+)
+
+type stubVideoSlot struct{}
+
+func (stubVideoSlot) Run(context.Context, FeedConfig, <-chan FeedConfig) error {
+	return nil
+}
+
+type stubAudioSlot struct{}
+
+func (stubAudioSlot) Run(context.Context, FeedConfig, <-chan FeedConfig) error {
+	return nil
+}
+
+type stubSyncSlot struct{}
+
+func (stubSyncSlot) Run(context.Context, SyncPairConfig, <-chan SyncPairConfig) error {
+	return nil
+}
+
+func TestNewSessionControllerValidatesSlots(t *testing.T) {
+	video := stubVideoSlot{}
+	audio := stubAudioSlot{}
+	sync := stubSyncSlot{}
+	tests := []struct {
+		name  string
+		video VideoSlotRunner
+		audio AudioSlotRunner
+		sync  SyncSlotRunner
+		want  error
+	}{
+		{"missing video", nil, audio, sync, ErrVideoSlotRequired},
+		{"missing audio", video, nil, sync, ErrAudioSlotRequired},
+		{"missing sync", video, audio, nil, ErrSyncSlotRequired},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			controller, err := NewSessionController(tt.video, tt.audio, tt.sync, nil)
+			if controller != nil {
+				t.Fatalf("NewSessionController() controller = %#v, want nil", controller)
+			}
+			if !errors.Is(err, tt.want) {
+				t.Fatalf("NewSessionController() error = %v, want %v", err, tt.want)
+			}
+		})
+	}
+}
+
+func TestNewSessionControllerAllowsUnavailableSynchronization(t *testing.T) {
+	video := stubVideoSlot{}
+	audio := stubAudioSlot{}
+	sync := stubSyncSlot{}
+	controller, err := NewSessionController(video, audio, sync, nil)
+	if err != nil {
+		t.Fatalf("NewSessionController() error = %v", err)
+	}
+	if controller == nil {
+		t.Fatal("NewSessionController() controller is nil")
+	}
+	if controller.videoSlot != video || controller.audioSlot != audio || controller.syncSlot != sync {
+		t.Fatalf("NewSessionController() = %#v", controller)
+	}
+	if controller.canSync != nil {
+		t.Fatal("nil sync predicate was not preserved")
+	}
+}
+
+func TestNewSessionControllerStoresSyncPredicate(t *testing.T) {
+	predicate := func(FeedConfig, FeedConfig) bool { return true }
+	controller, err := NewSessionController(
+		stubVideoSlot{}, stubAudioSlot{}, stubSyncSlot{}, predicate,
+	)
+	if err != nil {
+		t.Fatalf("NewSessionController() error = %v", err)
+	}
+	if controller.canSync == nil || !controller.canSync(FeedConfig{}, FeedConfig{}) {
+		t.Fatal("sync predicate was not stored")
+	}
+}
+
+type controllerEvent struct {
+	unit   Unit
+	action string
+	feed   FeedConfig
+	pair   SyncPairConfig
+}
+
+type recordingVideoSlot struct{ events chan<- controllerEvent }
+
+func (s recordingVideoSlot) Run(
+	ctx context.Context,
+	initial FeedConfig,
+	commands <-chan FeedConfig,
+) error {
+	s.events <- controllerEvent{unit: UnitVideo, action: "start", feed: initial}
+	for {
+		select {
+		case config := <-commands:
+			s.events <- controllerEvent{unit: UnitVideo, action: "command", feed: config}
+		case <-ctx.Done():
+			s.events <- controllerEvent{unit: UnitVideo, action: "stop"}
+			return ctx.Err()
+		}
+	}
+}
+
+type recordingAudioSlot struct{ events chan<- controllerEvent }
+
+func (s recordingAudioSlot) Run(
+	ctx context.Context,
+	initial FeedConfig,
+	commands <-chan FeedConfig,
+) error {
+	s.events <- controllerEvent{unit: UnitAudio, action: "start", feed: initial}
+	for {
+		select {
+		case config := <-commands:
+			s.events <- controllerEvent{unit: UnitAudio, action: "command", feed: config}
+		case <-ctx.Done():
+			s.events <- controllerEvent{unit: UnitAudio, action: "stop"}
+			return ctx.Err()
+		}
+	}
+}
+
+type recordingSyncSlot struct{ events chan<- controllerEvent }
+
+func (s recordingSyncSlot) Run(
+	ctx context.Context,
+	initial SyncPairConfig,
+	commands <-chan SyncPairConfig,
+) error {
+	s.events <- controllerEvent{unit: UnitSync, action: "start", pair: initial}
+	for {
+		select {
+		case config := <-commands:
+			s.events <- controllerEvent{unit: UnitSync, action: "command", pair: config}
+		case <-ctx.Done():
+			s.events <- controllerEvent{unit: UnitSync, action: "stop"}
+			return ctx.Err()
+		}
+	}
+}
+
+func newRecordingController(t *testing.T, events chan<- controllerEvent) *SessionController {
+	t.Helper()
+	controller, err := NewSessionController(
+		recordingVideoSlot{events},
+		recordingAudioSlot{events},
+		recordingSyncSlot{events},
+		func(FeedConfig, FeedConfig) bool { return true },
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return controller
+}
+
+func receiveControllerEvent(t *testing.T, events <-chan controllerEvent) controllerEvent {
+	t.Helper()
+	select {
+	case event := <-events:
+		return event
+	case <-time.After(time.Second):
+		t.Fatal("controller event timed out")
+		return controllerEvent{}
+	}
+}
+
+func receiveIndependentStarts(t *testing.T, events <-chan controllerEvent) {
+	t.Helper()
+	seen := map[Unit]bool{}
+	for len(seen) < 2 {
+		event := receiveControllerEvent(t, events)
+		if event.action != "start" || (event.unit != UnitVideo && event.unit != UnitAudio) {
+			t.Fatalf("unexpected initial event: %+v", event)
+		}
+		seen[event.unit] = true
+	}
+}
+
+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)
+	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)
+
+	want := FeedConfig{Domain: "/new-video", UUID: "new-video", Active: true}
+	commands <- SessionCommand{Kind: CommandSetVideo, Config: want}
+	event := receiveControllerEvent(t, events)
+	if event.unit != UnitVideo || event.action != "command" || event.feed != want {
+		t.Fatalf("replacement event = %+v", event)
+	}
+	select {
+	case event := <-events:
+		t.Fatalf("unchanged audio slot was disturbed: %+v", event)
+	case <-time.After(20 * time.Millisecond):
+	}
+
+	close(commands)
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatalf("Run() error = %v, want nil", err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("Run() did not stop")
+	}
+}
+
+func TestSessionControllerStopsIndependentSlotsBeforeStartingSync(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)
+
+	commands <- SessionCommand{Kind: CommandEnableSync}
+	stopped := map[Unit]bool{}
+	for {
+		event := receiveControllerEvent(t, events)
+		if event.unit == UnitSync && event.action == "start" {
+			if !stopped[UnitVideo] || !stopped[UnitAudio] {
+				t.Fatalf("sync started before both independent slots stopped: %v", stopped)
+			}
+			break
+		}
+		if event.action != "stop" || (event.unit != UnitVideo && event.unit != UnitAudio) {
+			t.Fatalf("unexpected transition event: %+v", event)
+		}
+		stopped[event.unit] = true
+	}
+
+	close(commands)
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatalf("Run() error = %v, want nil", err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("Run() did not stop")
+	}
+}
+
+func TestSessionControllerCancellationStopsAndJoinsRuntime(t *testing.T) {
+	events := make(chan controllerEvent, 32)
+	controller := newRecordingController(t, events)
+	initial := validCommandSession()
+	initial.SyncRequested = false
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() { done <- controller.Run(ctx, initial, make(chan SessionCommand)) }()
+	receiveIndependentStarts(t, events)
+	cancel()
+
+	select {
+	case err := <-done:
+		if !errors.Is(err, context.Canceled) {
+			t.Fatalf("Run() error = %v, want context.Canceled", err)
+		}
+	case <-time.After(time.Second):
+		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)
+	}
+}

internal/playback/session_plan.go

@@ -0,0 +1,76 @@
+package playback
+
+import "fmt"
+
+type SessionTopology uint8
+
+const (
+	TopologyIdle SessionTopology = iota
+	TopologyIndependent
+	TopologySynchronized
+)
+
+type SyncPredicate func(video, audio FeedConfig) bool
+
+type SessionPlan struct {
+	Topology SessionTopology
+
+	// Desired states for independent slots.
+	Video FeedConfig
+	Audio FeedConfig
+
+	// Desired state for the synchronized slot.
+	Sync SyncPairConfig
+}
+
+func BuildSessionPlan(
+	desired SessionConfig,
+	canSync SyncPredicate,
+) (SessionPlan, error) {
+	if err := desired.Validate(); err != nil {
+		return SessionPlan{}, fmt.Errorf("validate desired session: %w", err)
+	}
+
+	if desired.SyncRequested &&
+		desired.Video.Active &&
+		desired.Audio.Active &&
+		canSync != nil &&
+		canSync(desired.Video, desired.Audio) {
+		return SessionPlan{
+			Topology: TopologySynchronized,
+			Video:    stoppedFeed(desired.Video),
+			Audio:    stoppedFeed(desired.Audio),
+			Sync: SyncPairConfig{
+				Video: desired.Video,
+				Audio: desired.Audio,
+			},
+		}, nil
+	}
+
+	if desired.Video.Active || desired.Audio.Active {
+		return SessionPlan{
+			Topology: TopologyIndependent,
+			Video:    desired.Video,
+			Audio:    desired.Audio,
+			Sync: SyncPairConfig{
+				Video: stoppedFeed(desired.Video),
+				Audio: stoppedFeed(desired.Audio),
+			},
+		}, nil
+	}
+
+	return SessionPlan{
+		Topology: TopologyIdle,
+		Video:    stoppedFeed(desired.Video),
+		Audio:    stoppedFeed(desired.Audio),
+		Sync: SyncPairConfig{
+			Video: stoppedFeed(desired.Video),
+			Audio: stoppedFeed(desired.Audio),
+		},
+	}, nil
+}
+
+func stoppedFeed(config FeedConfig) FeedConfig {
+	config.Active = false
+	return config
+}

internal/playback/session_plan_test.go

@@ -0,0 +1,130 @@
+package playback
+
+import (
+	"errors"
+	"testing"
+)
+
+func TestBuildSessionPlan(t *testing.T) {
+	base := validCommandSession()
+	stoppedVideo := stoppedFeed(base.Video)
+	stoppedAudio := stoppedFeed(base.Audio)
+	independent := func(video, audio FeedConfig) SessionPlan {
+		return SessionPlan{
+			Topology: TopologyIndependent,
+			Video:    video,
+			Audio:    audio,
+			Sync:     SyncPairConfig{Video: stoppedFeed(video), Audio: stoppedFeed(audio)},
+		}
+	}
+	idle := func(video, audio FeedConfig) SessionPlan {
+		return SessionPlan{
+			Topology: TopologyIdle,
+			Video:    stoppedFeed(video),
+			Audio:    stoppedFeed(audio),
+			Sync:     SyncPairConfig{Video: stoppedFeed(video), Audio: stoppedFeed(audio)},
+		}
+	}
+
+	tests := []struct {
+		name    string
+		desired SessionConfig
+		canSync SyncPredicate
+		want    SessionPlan
+	}{
+		{
+			name:    "both active with sync disabled stay independent",
+			desired: func() SessionConfig { c := base; c.SyncRequested = false; return c }(),
+			canSync: func(FeedConfig, FeedConfig) bool { return true },
+			want:    independent(base.Video, base.Audio),
+		},
+		{
+			name:    "eligible requested pair is synchronized",
+			desired: base,
+			canSync: func(FeedConfig, FeedConfig) bool { return true },
+			want: SessionPlan{
+				Topology: TopologySynchronized,
+				Video:    stoppedVideo,
+				Audio:    stoppedAudio,
+				Sync:     SyncPairConfig{Video: base.Video, Audio: base.Audio},
+			},
+		},
+		{
+			name:    "ineligible requested pair remains independent",
+			desired: base,
+			canSync: func(FeedConfig, FeedConfig) bool { return false },
+			want:    independent(base.Video, base.Audio),
+		},
+		{
+			name:    "nil capability remains independent",
+			desired: base,
+			want:    independent(base.Video, base.Audio),
+		},
+		{
+			name:    "video only",
+			desired: func() SessionConfig { c := base; c.Audio.Active = false; return c }(),
+			canSync: func(FeedConfig, FeedConfig) bool { return true },
+			want:    independent(base.Video, stoppedAudio),
+		},
+		{
+			name:    "audio only",
+			desired: func() SessionConfig { c := base; c.Video.Active = false; return c }(),
+			canSync: func(FeedConfig, FeedConfig) bool { return true },
+			want:    independent(stoppedVideo, base.Audio),
+		},
+		{
+			name:    "stopped feeds are idle and retain configuration",
+			desired: func() SessionConfig { c := base; c.Video.Active = false; c.Audio.Active = false; return c }(),
+			canSync: func(FeedConfig, FeedConfig) bool { return true },
+			want:    idle(stoppedVideo, stoppedAudio),
+		},
+		{
+			name:    "empty session is idle",
+			desired: func() SessionConfig { c := base; c.Video = FeedConfig{}; c.Audio = FeedConfig{}; return c }(),
+			canSync: func(FeedConfig, FeedConfig) bool { return true },
+			want:    idle(FeedConfig{}, FeedConfig{}),
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := BuildSessionPlan(tt.desired, tt.canSync)
+			if err != nil {
+				t.Fatalf("BuildSessionPlan() error = %v", err)
+			}
+			if got != tt.want {
+				t.Fatalf("BuildSessionPlan() = %#v, want %#v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestBuildSessionPlanRejectsInvalidSession(t *testing.T) {
+	desired := validCommandSession()
+	desired.Video = FeedConfig{UUID: "video", Active: true}
+	got, err := BuildSessionPlan(desired, nil)
+	if !errors.Is(err, ErrFeedDomainRequired) {
+		t.Fatalf("BuildSessionPlan() error = %v, want %v", err, ErrFeedDomainRequired)
+	}
+	if got != (SessionPlan{}) {
+		t.Fatalf("BuildSessionPlan() = %#v, want zero plan", got)
+	}
+}
+
+func TestBuildSessionPlanPassesCompleteFeedsToPredicate(t *testing.T) {
+	desired := validCommandSession()
+	called := false
+	_, err := BuildSessionPlan(desired, func(video, audio FeedConfig) bool {
+		called = true
+		if video != desired.Video || audio != desired.Audio {
+			t.Fatalf("predicate feeds = %#v %#v", video, audio)
+		}
+		return true
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !called {
+		t.Fatal("sync predicate was not called")
+	}
+}

internal/source/source.go

@@ -598,8 +598,10 @@ func (s *SyncSource) Close() error {
 	return s.inst.Close()
 }
 
-// NextSync reads both at a synced timestamp. Returns video Frame + audio AudioFrame
+// NextSync reads one video frame and the audio interval between this video
-func (s *SyncSource) NextSync(ctx context.Context, audioBatch uint64, timeout time.Duration) (Frame, AudioFrame, error) {
+// timestamp and the next. Deriving the interval for every frame preserves
+// exact long-term timing for fractional video rates.
+func (s *SyncSource) NextSync(ctx context.Context, timeout time.Duration) (Frame, AudioFrame, error) {
 	var timeouts int
 	for {
 		select {
@@ -635,7 +637,30 @@ func (s *SyncSource) NextSync(ctx context.Context, audioBatch uint64, timeout ti
 			}
 			// read audio at the same timestamp
 			aIdx := mxl.TimestampToIndex(s.aRate, ts)
+			nextTimestamp := mxl.IndexToTimestamp(s.rate, s.idx+1)
+			nextAudioIndex := mxl.TimestampToIndex(s.aRate, nextTimestamp)
+			if nextAudioIndex <= aIdx {
+				return Frame{}, AudioFrame{}, wrapError(
+					"calculate synchronized audio interval",
+					ErrorKindInvalidConfig,
+					fmt.Errorf("invalid audio interval: %d..%d", aIdx, nextAudioIndex),
+				)
+			}
+			audioBatch := nextAudioIndex - aIdx
 			av, aerr := s.ar.GetSamples(aIdx, int(audioBatch), 50*time.Millisecond)
+			if aerr != nil {
+				kind := ErrorKindUnavailable
+				if errors.Is(aerr, mxl.ErrTimeout) ||
+					errors.Is(aerr, mxl.ErrOutOfRangeEarly) ||
+					errors.Is(aerr, mxl.ErrOutOfRangeLate) {
+					kind = ErrorKindTemporary
+				}
+				return Frame{}, AudioFrame{}, wrapError(
+					"read synchronized audio",
+					kind,
+					aerr,
+				)
+			}
 			vFrame := Frame{
 				Index: g.Index, Width: s.width, Height: s.height,
 				Stride: s.stride, Size: g.GrainSize,
@@ -643,8 +668,6 @@ func (s *SyncSource) NextSync(ctx context.Context, audioBatch uint64, timeout ti
 			}
 			s.idx++
 
-			var aFrame AudioFrame
-			if aerr == nil {
 			samples := make([][]byte, s.chans)
 			for ch := uint64(0); ch < s.chans; ch++ {
 				f1, f2, _ := av.ChannelFragments(ch)
@@ -654,12 +677,10 @@ func (s *SyncSource) NextSync(ctx context.Context, audioBatch uint64, timeout ti
 					samples[ch] = f1
 				}
 			}
-				aFrame = AudioFrame{
+			aFrame := AudioFrame{
 				Index: aIdx, SampleCount: audioBatch,
 				Channels: s.chans, Samples: samples,
 			}
-			}
-			// even if audio failed, video returns
 			return vFrame, aFrame, nil
 		case errors.Is(err, mxl.ErrTimeout), errors.Is(err, mxl.ErrOutOfRangeEarly), errors.Is(err, mxl.ErrOutOfRangeLate):
 			timeouts++