session command

imgui.ini

@@ -14,7 +14,7 @@ Size=200,200
 Collapsed=0
 
 [Window][Connection]
-Pos=177,324
+Pos=475,529
-Size=640,354
+Size=640,352
 Collapsed=0
 

internal/adapter/mxl/sync.go

@@ -0,0 +1,175 @@
+package mxladapter
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"mxl-player/internal/playback"
+	"mxl-player/internal/source"
+
+	mxl "github.com/qvest-digital/go-mxl/mxl"
+)
+
+const (
+	DefaultSyncReadTimeout   = 200 * time.Millisecond
+	DefaultSyncBatchDuration = 10 * time.Millisecond
+)
+
+var ErrNativeSyncDifferentDomains = errors.New(
+	"native MXL synchronization requires matching domains",
+)
+
+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)
+
+	AudioRate() mxl.Rational
+	Close() error
+}
+
+type syncReader struct {
+	source          localSyncSource
+	readTimeout     time.Duration
+	audioBatch      uint64
+	rateNumerator   int64
+	rateDenominator int64
+}
+
+var _ playback.SyncReaderFactory = SyncFactory{}
+var _ playback.SyncReader = (*syncReader)(nil)
+
+func (f SyncFactory) OpenSync(
+	ctx context.Context,
+	videoConfig playback.FeedConfig,
+	audioConfig playback.FeedConfig,
+) (playback.SyncReader, error) {
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
+	if err := videoConfig.Validate(); err != nil {
+		return nil, &source.SourceError{
+			Op:   "validate sync video feed",
+			Kind: source.ErrorKindInvalidConfig,
+			Err:  err,
+		}
+	}
+	if err := audioConfig.Validate(); err != nil {
+		return nil, &source.SourceError{
+			Op:   "validate sync audio feed",
+			Kind: source.ErrorKindInvalidConfig,
+			Err:  err,
+		}
+	}
+	if !videoConfig.IsConfigured() {
+		return nil, &source.SourceError{
+			Op:   "validate sync video feed",
+			Kind: source.ErrorKindInvalidConfig,
+			Err:  errors.New("sync video feed is not configured"),
+		}
+	}
+	if !audioConfig.IsConfigured() {
+		return nil, &source.SourceError{
+			Op:   "validate sync audio feed",
+			Kind: source.ErrorKindInvalidConfig,
+			Err:  errors.New("sync audio feed is not configured"),
+		}
+	}
+
+	if videoConfig.Domain != audioConfig.Domain {
+		return nil, &source.SourceError{
+			Op:   "validate native MXL sync group",
+			Kind: source.ErrorKindInvalidConfig,
+			Err:  ErrNativeSyncDifferentDomains,
+		}
+	}
+
+	open := f.open
+	if open == nil {
+		open = func(domain, videoUUID, audioUUID string) (localSyncSource, error) {
+			return source.OpenSameDomainSync(domain, videoUUID, audioUUID)
+		}
+	}
+	src, err := open(
+		videoConfig.Domain,
+		videoConfig.UUID,
+		audioConfig.UUID,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("open native MXL sync group: %w", err)
+	}
+
+	if err := ctx.Err(); err != nil {
+		_ = src.Close()
+		return nil, err
+	}
+
+	readTimeout := f.ReadTimeout
+	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
+}
+
+func (r *syncReader) ReadSync(
+	ctx context.Context,
+) (playback.SyncFrame, error) {
+	video, audio, err := r.source.NextSync(ctx, r.audioBatch, r.readTimeout)
+	if err != nil {
+		return playback.SyncFrame{}, err
+	}
+	return playback.SyncFrame{
+		Video: playback.VideoFrame{
+			Index:   video.Index,
+			Width:   video.Width,
+			Height:  video.Height,
+			Stride:  video.Stride,
+			Size:    video.Size,
+			Invalid: video.Invalid,
+			Payload: video.Payload,
+		},
+		Audio: playback.AudioFrame{
+			Index:                 audio.Index,
+			SampleCount:           audio.SampleCount,
+			Channels:              audio.Channels,
+			SampleRateNumerator:   r.rateNumerator,
+			SampleRateDenominator: r.rateDenominator,
+			Samples:               audio.Samples,
+		},
+	}, nil
+}
+
+func (r *syncReader) Close() error {
+	return r.source.Close()
+}

internal/adapter/mxl/sync_test.go

@@ -0,0 +1,143 @@
+package mxladapter
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"mxl-player/internal/playback"
+	"mxl-player/internal/source"
+
+	mxl "github.com/qvest-digital/go-mxl/mxl"
+)
+
+type fakeLocalSyncSource struct {
+	video      source.Frame
+	audio      source.AudioFrame
+	readErr    error
+	rate       mxl.Rational
+	batch      uint64
+	timeout    time.Duration
+	closed     bool
+	closeError error
+}
+
+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
+}
+
+func (s *fakeLocalSyncSource) AudioRate() mxl.Rational { return s.rate }
+func (s *fakeLocalSyncSource) Close() error {
+	s.closed = true
+	return s.closeError
+}
+
+func syncFeedConfigs() (playback.FeedConfig, playback.FeedConfig) {
+	return playback.FeedConfig{Domain: "/mxl", UUID: "video", Active: true},
+		playback.FeedConfig{Domain: "/mxl", UUID: "audio", Active: true}
+}
+
+func TestSyncFactoryRejectsDifferentDomains(t *testing.T) {
+	video, audio := syncFeedConfigs()
+	audio.Domain = "/other"
+	reader, err := (SyncFactory{}).OpenSync(context.Background(), video, audio)
+	if reader != nil {
+		t.Fatal("OpenSync() reader is not nil")
+	}
+	if !errors.Is(err, ErrNativeSyncDifferentDomains) {
+		t.Fatalf("OpenSync() error = %v, want %v", err, ErrNativeSyncDifferentDomains)
+	}
+	if source.KindOf(err) != source.ErrorKindInvalidConfig {
+		t.Fatalf("error kind = %v, want invalid config", source.KindOf(err))
+	}
+	if ShouldRetry(err) {
+		t.Fatal("ShouldRetry() = true for different domains")
+	}
+}
+
+func TestSyncFactoryUsesDefaultsAndForwardsFeeds(t *testing.T) {
+	fake := &fakeLocalSyncSource{rate: mxl.Rational{Num: 48_000, Den: 1}}
+	var domain, videoUUID, audioUUID string
+	factory := SyncFactory{open: func(d, v, a string) (localSyncSource, error) {
+		domain, videoUUID, audioUUID = d, v, a
+		return fake, nil
+	}}
+	video, audio := syncFeedConfigs()
+
+	reader, err := factory.OpenSync(context.Background(), video, audio)
+	if err != nil {
+		t.Fatalf("OpenSync() error = %v", err)
+	}
+	if domain != video.Domain || videoUUID != video.UUID || audioUUID != audio.UUID {
+		t.Fatalf("open args = %q %q %q", domain, videoUUID, audioUUID)
+	}
+	got := reader.(*syncReader)
+	if got.readTimeout != DefaultSyncReadTimeout || got.audioBatch != 480 {
+		t.Fatalf("reader timeout=%s batch=%d, want %s and 480", got.readTimeout, got.audioBatch, 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)
+	}
+}
+
+func TestSyncReaderConvertsPairWithoutCopying(t *testing.T) {
+	videoPayload := []byte{1, 2, 3}
+	audioSamples := [][]byte{{4, 5, 6, 7}}
+	fake := &fakeLocalSyncSource{
+		video: source.Frame{Index: 10, Width: 20, Height: 30, Payload: videoPayload},
+		audio: source.AudioFrame{Index: 40, SampleCount: 1, Channels: 1, Samples: audioSamples},
+		rate:  mxl.Rational{Num: 48_000, Den: 1},
+	}
+	reader := &syncReader{
+		source: fake, readTimeout: 7 * time.Millisecond, audioBatch: 12,
+		rateNumerator: 48_000, rateDenominator: 1,
+	}
+
+	frame, err := reader.ReadSync(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if fake.batch != 12 || fake.timeout != 7*time.Millisecond {
+		t.Fatalf("NextSync() batch=%d timeout=%s", fake.batch, fake.timeout)
+	}
+	if frame.Video.Index != 10 || frame.Audio.Index != 40 || frame.Audio.SampleRateNumerator != 48_000 {
+		t.Fatalf("frame = %+v", frame)
+	}
+	if &frame.Video.Payload[0] != &videoPayload[0] || &frame.Audio.Samples[0][0] != &audioSamples[0][0] {
+		t.Fatal("sync payload was copied")
+	}
+}
+
+func TestSyncFactoryReturnsPreCanceledContextWithoutOpening(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	opened := false
+	factory := SyncFactory{open: func(string, string, string) (localSyncSource, error) {
+		opened = true
+		return nil, nil
+	}}
+	video, audio := syncFeedConfigs()
+	reader, err := factory.OpenSync(ctx, video, audio)
+	if reader != nil || !errors.Is(err, context.Canceled) || opened {
+		t.Fatalf("reader=%v error=%v opened=%t", reader, err, opened)
+	}
+}

internal/playback/session_command.go

@@ -0,0 +1,110 @@
+package playback
+
+import (
+	"errors"
+	"fmt"
+)
+
+type SessionCommandKind uint8
+
+const (
+	CommandSetVideo SessionCommandKind = iota + 1
+	CommandSetAudio
+	CommandStopVideo
+	CommandStopAudio
+	CommandStopAll
+	CommandResumeVideo
+	CommandResumeAudio
+	CommandResumeAll
+	CommandRemoveVideo
+	CommandRemoveAudio
+	CommandEnableSync
+	CommandDisableSync
+)
+
+type SessionCommand struct {
+	Kind   SessionCommandKind
+	Config FeedConfig // Used only by SetVideo and SetAudio.
+}
+
+var (
+	ErrUnknownSessionCommand = errors.New("unknown session command")
+	ErrVideoNotConfigured    = errors.New("video feed is not configured")
+	ErrAudioNotConfigured    = errors.New("audio feed is not configured")
+)
+
+func ApplySessionCommand(
+	current SessionConfig,
+	command SessionCommand,
+) (SessionConfig, error) {
+	next := current
+	switch command.Kind {
+	case CommandSetVideo:
+		if !command.Config.IsConfigured() {
+			return current, ErrVideoNotConfigured
+		}
+		if err := command.Config.Validate(); err != nil {
+			return current, fmt.Errorf("validate video command: %w", err)
+		}
+		next.Video = command.Config
+
+	case CommandSetAudio:
+		if !command.Config.IsConfigured() {
+			return current, ErrAudioNotConfigured
+		}
+		if err := command.Config.Validate(); err != nil {
+			return current, fmt.Errorf("validate audio command: %w", err)
+		}
+		next.Audio = command.Config
+
+	case CommandStopVideo:
+		next.Video.Active = false
+		next.SyncRequested = false
+
+	case CommandStopAudio:
+		next.Audio.Active = false
+		next.SyncRequested = false
+
+	case CommandStopAll:
+		next.Video.Active = false
+		next.Audio.Active = false
+
+	case CommandResumeVideo:
+		if !next.Video.IsConfigured() {
+			return current, ErrVideoNotConfigured
+		}
+		next.Video.Active = true
+
+	case CommandResumeAudio:
+		if !next.Audio.IsConfigured() {
+			return current, ErrAudioNotConfigured
+		}
+		next.Audio.Active = true
+
+	case CommandResumeAll:
+		next.Video.Active = next.Video.IsConfigured()
+		next.Audio.Active = next.Audio.IsConfigured()
+
+	case CommandRemoveVideo:
+		next.Video = FeedConfig{}
+		next.SyncRequested = false
+
+	case CommandRemoveAudio:
+		next.Audio = FeedConfig{}
+		next.SyncRequested = false
+
+	case CommandEnableSync:
+		next.SyncRequested = true
+
+	case CommandDisableSync:
+		next.SyncRequested = false
+
+	default:
+		return current, ErrUnknownSessionCommand
+	}
+
+	if err := next.Validate(); err != nil {
+		return current, fmt.Errorf("validate desired session: %w", err)
+	}
+	return next, nil
+}

internal/playback/session_command_test.go

@@ -0,0 +1,185 @@
+package playback
+
+import (
+	"errors"
+	"testing"
+	"time"
+)
+
+func validCommandSession() SessionConfig {
+	return SessionConfig{
+		Video:         FeedConfig{Domain: "/video", UUID: "video", Active: true},
+		Audio:         FeedConfig{Domain: "/audio", UUID: "audio", Active: true},
+		SyncRequested: true,
+		Retry: RetryPolicy{
+			MaxAttempts:  3,
+			InitialDelay: time.Millisecond,
+			MaxDelay:     time.Second,
+		},
+	}
+}
+
+func TestApplySessionCommand(t *testing.T) {
+	base := validCommandSession()
+	newVideo := FeedConfig{Domain: "/new-video", UUID: "new-video", Active: false}
+	newAudio := FeedConfig{Domain: "/new-audio", UUID: "new-audio", Active: true}
+
+	tests := []struct {
+		name    string
+		current SessionConfig
+		command SessionCommand
+		want    SessionConfig
+	}{
+		{
+			name:    "set video replaces complete config",
+			current: base,
+			command: SessionCommand{Kind: CommandSetVideo, Config: newVideo},
+			want:    func() SessionConfig { c := base; c.Video = newVideo; return c }(),
+		},
+		{
+			name:    "set audio replaces complete config",
+			current: base,
+			command: SessionCommand{Kind: CommandSetAudio, Config: newAudio},
+			want:    func() SessionConfig { c := base; c.Audio = newAudio; return c }(),
+		},
+		{
+			name:    "stop video disables sync",
+			current: base,
+			command: SessionCommand{Kind: CommandStopVideo},
+			want:    func() SessionConfig { c := base; c.Video.Active = false; c.SyncRequested = false; return c }(),
+		},
+		{
+			name:    "stop audio disables sync",
+			current: base,
+			command: SessionCommand{Kind: CommandStopAudio},
+			want:    func() SessionConfig { c := base; c.Audio.Active = false; c.SyncRequested = false; return c }(),
+		},
+		{
+			name:    "stop all preserves sync request",
+			current: base,
+			command: SessionCommand{Kind: CommandStopAll},
+			want:    func() SessionConfig { c := base; c.Video.Active = false; c.Audio.Active = false; return c }(),
+		},
+		{
+			name:    "resume video",
+			current: func() SessionConfig { c := base; c.Video.Active = false; return c }(),
+			command: SessionCommand{Kind: CommandResumeVideo},
+			want:    base,
+		},
+		{
+			name:    "resume audio",
+			current: func() SessionConfig { c := base; c.Audio.Active = false; return c }(),
+			command: SessionCommand{Kind: CommandResumeAudio},
+			want:    base,
+		},
+		{
+			name:    "resume all activates only configured feeds",
+			current: func() SessionConfig { c := base; c.Video.Active = false; c.Audio = FeedConfig{}; return c }(),
+			command: SessionCommand{Kind: CommandResumeAll},
+			want:    func() SessionConfig { c := base; c.Audio = FeedConfig{}; return c }(),
+		},
+		{
+			name:    "remove video clears config and disables sync",
+			current: base,
+			command: SessionCommand{Kind: CommandRemoveVideo},
+			want:    func() SessionConfig { c := base; c.Video = FeedConfig{}; c.SyncRequested = false; return c }(),
+		},
+		{
+			name:    "remove audio clears config and disables sync",
+			current: base,
+			command: SessionCommand{Kind: CommandRemoveAudio},
+			want:    func() SessionConfig { c := base; c.Audio = FeedConfig{}; c.SyncRequested = false; return c }(),
+		},
+		{
+			name: "enable sync without feeds records request",
+			current: func() SessionConfig {
+				c := base
+				c.Video = FeedConfig{}
+				c.Audio = FeedConfig{}
+				c.SyncRequested = false
+				return c
+			}(),
+			command: SessionCommand{Kind: CommandEnableSync},
+			want: func() SessionConfig {
+				c := base
+				c.Video = FeedConfig{}
+				c.Audio = FeedConfig{}
+				c.SyncRequested = true
+				return c
+			}(),
+		},
+		{
+			name:    "disable sync",
+			current: base,
+			command: SessionCommand{Kind: CommandDisableSync},
+			want:    func() SessionConfig { c := base; c.SyncRequested = false; return c }(),
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := ApplySessionCommand(tt.current, tt.command)
+			if err != nil {
+				t.Fatalf("ApplySessionCommand() error = %v", err)
+			}
+			if got != tt.want {
+				t.Fatalf("ApplySessionCommand() = %#v, want %#v", got, tt.want)
+			}
+			if got.Retry != tt.current.Retry {
+				t.Fatalf("retry changed from %#v to %#v", tt.current.Retry, got.Retry)
+			}
+		})
+	}
+}
+
+func TestApplySessionCommandFailurePreservesState(t *testing.T) {
+	base := validCommandSession()
+	tests := []struct {
+		name    string
+		current SessionConfig
+		command SessionCommand
+		wantErr error
+	}{
+		{"set empty video", base, SessionCommand{Kind: CommandSetVideo}, ErrVideoNotConfigured},
+		{"set empty audio", base, SessionCommand{Kind: CommandSetAudio}, ErrAudioNotConfigured},
+		{"invalid video", base, SessionCommand{Kind: CommandSetVideo, Config: FeedConfig{UUID: "video"}}, ErrFeedDomainRequired},
+		{"invalid audio", base, SessionCommand{Kind: CommandSetAudio, Config: FeedConfig{UUID: "audio"}}, ErrFeedDomainRequired},
+		{
+			"resume missing video",
+			func() SessionConfig { c := base; c.Video = FeedConfig{}; return c }(),
+			SessionCommand{Kind: CommandResumeVideo},
+			ErrVideoNotConfigured,
+		},
+		{
+			"resume missing audio",
+			func() SessionConfig { c := base; c.Audio = FeedConfig{}; return c }(),
+			SessionCommand{Kind: CommandResumeAudio},
+			ErrAudioNotConfigured,
+		},
+		{"unknown command", base, SessionCommand{Kind: 255}, ErrUnknownSessionCommand},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := ApplySessionCommand(tt.current, tt.command)
+			if !errors.Is(err, tt.wantErr) {
+				t.Fatalf("ApplySessionCommand() error = %v, want %v", err, tt.wantErr)
+			}
+			if got != tt.current {
+				t.Fatalf("failed command changed state from %#v to %#v", tt.current, got)
+			}
+		})
+	}
+}
+
+func TestApplySessionCommandRejectsInvalidResult(t *testing.T) {
+	current := validCommandSession()
+	current.Retry = RetryPolicy{}
+	got, err := ApplySessionCommand(current, SessionCommand{Kind: CommandDisableSync})
+	if !errors.Is(err, ErrInvalidRetryDelay) {
+		t.Fatalf("ApplySessionCommand() error = %v, want %v", err, ErrInvalidRetryDelay)
+	}
+	if got != current {
+		t.Fatalf("failed command changed state from %#v to %#v", current, got)
+	}
+}

internal/playback/sync_slot.go

@@ -0,0 +1,122 @@
+package playback
+
+import (
+	"context"
+	"errors"
+	"fmt"
+)
+
+type SyncPairConfig struct {
+	Video FeedConfig
+	Audio FeedConfig
+}
+
+var (
+	ErrSyncWorkerRequired   = errors.New("sync worker is required")
+	ErrSyncActivityMismatch = errors.New(
+		"synchronized video and audio must have matching active states",
+	)
+)
+
+func (c SyncPairConfig) Validate() error {
+	if err := c.Video.Validate(); err != nil {
+		return fmt.Errorf("video: %w", err)
+	}
+	if err := c.Audio.Validate(); err != nil {
+		return fmt.Errorf("audio: %w", err)
+	}
+	if c.Video.Active != c.Audio.Active {
+		return ErrSyncActivityMismatch
+	}
+	return nil
+}
+
+func (c SyncPairConfig) Active() bool {
+	return c.Video.Active && c.Audio.Active
+}
+
+type SyncSlot struct {
+	worker *SyncWorker
+}
+
+func NewSyncSlot(worker *SyncWorker) (*SyncSlot, error) {
+	if worker == nil {
+		return nil, ErrSyncWorkerRequired
+	}
+	return &SyncSlot{
+		worker: worker,
+	}, nil
+}
+
+func (s *SyncSlot) Run(
+	ctx context.Context,
+	initial SyncPairConfig,
+	commands <-chan SyncPairConfig,
+) error {
+	if err := initial.Validate(); err != nil {
+		return fmt.Errorf("validate initial sync config: %w", err)
+	}
+
+	var (
+		workerCancel context.CancelFunc
+		workerDone   chan error
+	)
+
+	start := func(config SyncPairConfig) {
+		workerCtx, cancel := context.WithCancel(ctx)
+		done := make(chan error, 1)
+
+		workerCancel = cancel
+		workerDone = done
+
+		go func() {
+			done <- s.worker.Run(workerCtx, config.Video, config.Audio)
+		}()
+	}
+
+	stop := func() {
+		if workerCancel == nil {
+			return
+		}
+
+		workerCancel()
+		<-workerDone
+
+		workerCancel = nil
+		workerDone = nil
+	}
+
+	if initial.Active() {
+		start(initial)
+	}
+
+	for {
+		select {
+		case <-ctx.Done():
+			stop()
+			return ctx.Err()
+
+		case config, ok := <-commands:
+			if !ok {
+				stop()
+				return nil
+			}
+
+			if err := config.Validate(); err != nil {
+				// Ignore invalid commands without disturbing the current worker.
+				continue
+			}
+
+			stop()
+			if config.Active() {
+				start(config)
+			}
+
+		case <-workerDone:
+			// The worker stopped naturally or exhausted its retries.
+			workerCancel()
+			workerCancel = nil
+			workerDone = nil
+		}
+	}
+}

internal/playback/sync_slot_test.go

@@ -0,0 +1,231 @@
+package playback
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+)
+
+type slotSyncFactory struct {
+	opened chan SyncPairConfig
+
+	mu         sync.Mutex
+	active     int
+	maxActive  int
+	closeCount int
+}
+
+func newSlotSyncFactory() *slotSyncFactory {
+	return &slotSyncFactory{opened: make(chan SyncPairConfig, 8)}
+}
+
+func (f *slotSyncFactory) OpenSync(
+	_ context.Context,
+	video FeedConfig,
+	audio FeedConfig,
+) (SyncReader, error) {
+	f.mu.Lock()
+	f.active++
+	if f.active > f.maxActive {
+		f.maxActive = f.active
+	}
+	f.mu.Unlock()
+	f.opened <- SyncPairConfig{Video: video, Audio: audio}
+	return &slotSyncReader{factory: f}, nil
+}
+
+func (f *slotSyncFactory) counts() (active, maxActive, closeCount int) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.active, f.maxActive, f.closeCount
+}
+
+type slotSyncReader struct {
+	factory *slotSyncFactory
+}
+
+func (r *slotSyncReader) ReadSync(ctx context.Context) (SyncFrame, error) {
+	<-ctx.Done()
+	return SyncFrame{}, ctx.Err()
+}
+
+func (r *slotSyncReader) Close() error {
+	r.factory.mu.Lock()
+	defer r.factory.mu.Unlock()
+	r.factory.active--
+	r.factory.closeCount++
+	return nil
+}
+
+func newSlotTestSyncWorker(t *testing.T, factory SyncReaderFactory) *SyncWorker {
+	t.Helper()
+	worker, err := NewSyncWorker(
+		factory,
+		&fakeVideoSink{},
+		&fakeAudioSink{},
+		testRetryPolicy(1),
+		func(error) bool { return false },
+		nil,
+	)
+	if err != nil {
+		t.Fatalf("NewSyncWorker() error = %v", err)
+	}
+	return worker
+}
+
+func receiveSyncSlotOpen(t *testing.T, opened <-chan SyncPairConfig) SyncPairConfig {
+	t.Helper()
+	select {
+	case config := <-opened:
+		return config
+	case <-time.After(time.Second):
+		t.Fatal("sync worker did not open")
+		return SyncPairConfig{}
+	}
+}
+
+func testSyncPair(name string, active bool) SyncPairConfig {
+	return SyncPairConfig{
+		Video: FeedConfig{Domain: "/video", UUID: name + "-video", Active: active},
+		Audio: FeedConfig{Domain: "/audio", UUID: name + "-audio", Active: active},
+	}
+}
+
+func TestSyncPairConfigRejectsActivityMismatch(t *testing.T) {
+	config := testSyncPair("pair", true)
+	config.Audio.Active = false
+	if err := config.Validate(); !errors.Is(err, ErrSyncActivityMismatch) {
+		t.Fatalf("Validate() error = %v, want %v", err, ErrSyncActivityMismatch)
+	}
+}
+
+func TestNewSyncSlotRequiresWorker(t *testing.T) {
+	slot, err := NewSyncSlot(nil)
+	if slot != nil {
+		t.Fatalf("NewSyncSlot(nil) slot = %#v, want nil", slot)
+	}
+	if !errors.Is(err, ErrSyncWorkerRequired) {
+		t.Fatalf("NewSyncSlot(nil) error = %v, want %v", err, ErrSyncWorkerRequired)
+	}
+}
+
+func TestSyncSlotStartsInitialActivePair(t *testing.T) {
+	factory := newSlotSyncFactory()
+	slot, err := NewSyncSlot(newSlotTestSyncWorker(t, factory))
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	want := testSyncPair("first", true)
+	go func() { done <- slot.Run(ctx, want, make(chan SyncPairConfig)) }()
+
+	if got := receiveSyncSlotOpen(t, factory.opened); got != want {
+		t.Fatalf("opened config = %#v, want %#v", got, want)
+	}
+	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")
+	}
+	active, _, closed := factory.counts()
+	if active != 0 || closed != 1 {
+		t.Fatalf("reader counts = active %d, closed %d; want 0, 1", active, closed)
+	}
+}
+
+func TestSyncSlotReplacesWithoutOverlappingWorkers(t *testing.T) {
+	factory := newSlotSyncFactory()
+	slot, err := NewSyncSlot(newSlotTestSyncWorker(t, factory))
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	commands := make(chan SyncPairConfig)
+	done := make(chan error, 1)
+	first := testSyncPair("first", true)
+	second := testSyncPair("second", true)
+	go func() { done <- slot.Run(ctx, first, commands) }()
+	receiveSyncSlotOpen(t, factory.opened)
+	commands <- second
+	if got := receiveSyncSlotOpen(t, factory.opened); got != second {
+		t.Fatalf("replacement = %#v, want %#v", got, second)
+	}
+	cancel()
+	<-done
+	active, maxActive, closed := factory.counts()
+	if active != 0 || maxActive != 1 || closed != 2 {
+		t.Fatalf("counts = active %d, maximum %d, closed %d; want 0, 1, 2", active, maxActive, closed)
+	}
+}
+
+func TestSyncSlotIgnoresInvalidCommand(t *testing.T) {
+	factory := newSlotSyncFactory()
+	slot, err := NewSyncSlot(newSlotTestSyncWorker(t, factory))
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	commands := make(chan SyncPairConfig)
+	done := make(chan error, 1)
+	initial := testSyncPair("first", true)
+	go func() { done <- slot.Run(ctx, initial, commands) }()
+	receiveSyncSlotOpen(t, factory.opened)
+
+	invalid := testSyncPair("invalid", true)
+	invalid.Audio.Active = false
+	commands <- invalid
+	select {
+	case config := <-factory.opened:
+		t.Fatalf("invalid command opened config %#v", config)
+	case <-time.After(20 * time.Millisecond):
+	}
+	active, _, closed := factory.counts()
+	if active != 1 || closed != 0 {
+		t.Fatalf("invalid command disturbed reader: active %d, closed %d", active, closed)
+	}
+	cancel()
+	<-done
+}
+
+func TestSyncSlotInactivePairStopsWithoutRestart(t *testing.T) {
+	factory := newSlotSyncFactory()
+	slot, err := NewSyncSlot(newSlotTestSyncWorker(t, factory))
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	commands := make(chan SyncPairConfig)
+	done := make(chan error, 1)
+	go func() { done <- slot.Run(ctx, testSyncPair("first", true), commands) }()
+	receiveSyncSlotOpen(t, factory.opened)
+	commands <- testSyncPair("first", false)
+
+	deadline := time.Now().Add(time.Second)
+	for {
+		active, _, closed := factory.counts()
+		if active == 0 && closed == 1 {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatal("inactive pair did not stop reader")
+		}
+		time.Sleep(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 after commands closed")
+	}
+}