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 != ""
-			videoActive = videoStr != ""
+		audioActive = audioStr != ""
-			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,
-				}
+				},
-			}
+			})
-
+		}
-			audioConfig := playback.FeedConfig{}
+		if audioStr == "" {
-			if audioStr != "" {
+			enqueueCommand(playback.SessionCommand{Kind: playback.CommandRemoveAudio})
-				audioConfig = playback.FeedConfig{
+		} else {
+			enqueueCommand(playback.SessionCommand{
+				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{})
+	playbackDone := make(chan error, 1)
 	go func() {
-		defer close(playbackDone)
+		playbackDone <- player.Controller.Run(
-
+			ctx,
-		if useIndependentSlots {
+			args.playbackConfig(),
-			var slots sync.WaitGroup
+			player.Commands,
-			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)
-				}
-			}()
-
-			go func() {
-				defer slots.Done()
-
-				err := audioSlot.Run(
-					ctx,
-					playback.FeedConfig{
-						Domain: args.AudioDomain,
-						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,136 +527,109 @@ 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 {
-				if videoActive {
+				cimgui.Text("Video desired: active")
-					cimgui.Text("Video desired: active")
+			} else if videoStr != "" {
-				} else if videoStr != "" {
+				cimgui.Text("Video desired: stopped")
-					cimgui.Text("Video desired: stopped")
+			} else {
-				} else {
+				cimgui.Text("Video desired: not configured")
-					cimgui.Text("Video desired: not configured")
-				}
-
-				if status, ok := statusStore.Snapshot(playback.UnitVideo); ok {
-					cimgui.Text(fmt.Sprintf(
-						"Video actual: %s",
-						status.State,
-					))
-					cimgui.Text(fmt.Sprintf(
-						"Attempt: %d, failed: %d",
-						status.Attempt,
-						status.FailedAttempts,
-					))
-
-					if status.RetryIn > 0 {
-						cimgui.Text(fmt.Sprintf(
-							"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 useIndependentSlots && audioActive {
+
+			if status, ok := statusStore.Snapshot(playback.UnitVideo); ok {
+				cimgui.Text(fmt.Sprintf("Video actual: %s", status.State))
+				cimgui.Text(fmt.Sprintf("Attempt: %d, failed: %d", status.Attempt, status.FailedAttempts))
+				if status.RetryIn > 0 {
+					cimgui.Text(fmt.Sprintf("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 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 {
-				if audioActive {
+				cimgui.Text("Audio desired: active")
-					cimgui.Text("Audio desired: active")
+			} else if audioStr != "" {
-				} else if audioStr != "" {
+				cimgui.Text("Audio desired: stopped")
-					cimgui.Text("Audio desired: stopped")
+			} else {
-				} else {
+				cimgui.Text("Audio desired: not configured")
-					cimgui.Text("Audio desired: not configured")
+			}
+
+			if status, ok := statusStore.Snapshot(playback.UnitAudio); ok {
+				cimgui.Text(fmt.Sprintf("Audio actual: %s", status.State))
+				cimgui.Text(fmt.Sprintf("Attempt: %d, failed: %d", status.Attempt, status.FailedAttempts))
+				if status.RetryIn > 0 {
+					cimgui.Text(fmt.Sprintf("Retry in: %s", status.RetryIn.Round(time.Millisecond)))
 				}
-
+				if status.Err != nil {
-				if status, ok := statusStore.Snapshot(playback.UnitAudio); ok {
+					cimgui.TextWrapped(status.Err.Error())
-					cimgui.Text(fmt.Sprintf(
+				}
-						"Audio actual: %s",
+			} else {
-						status.State,
+				cimgui.Text("Audio actual: not started")
-					))
+			}
-					cimgui.Text(fmt.Sprintf(
+			if status, ok := statusStore.Snapshot(playback.UnitSync); ok {
-						"Attempt: %d, failed: %d",
+				cimgui.Text(fmt.Sprintf("Sync actual: %s", status.State))
-						status.Attempt,
+				if status.Err != nil {
-						status.FailedAttempts,
+					cimgui.TextWrapped(status.Err.Error())
-					))
-
-					if status.RetryIn > 0 {
-						cimgui.Text(fmt.Sprintf(
-							"Retry in: %s",
-							status.RetryIn.Round(time.Millisecond),
-						))
-					}
-
-					if status.Err != nil {
-						cimgui.TextWrapped(status.Err.Error())
-					}
-				} else {
-					cimgui.Text("Audio actual: not started")
 				}
 			}
 
@@ -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)
+	}
 }

imgui.ini

@@ -14,7 +14,7 @@ Size=200,200
 Collapsed=0
 
 [Window][Connection]
-Pos=1322,937
+Pos=322,387
-Size=613,164
+Size=618,275
 Collapsed=0
 

internal/adapter/mxl/sync.go

@@ -13,8 +13,7 @@ import (
 )
 
 const (
-	DefaultSyncReadTimeout   = 200 * time.Millisecond
+	DefaultSyncReadTimeout = 200 * time.Millisecond
-	DefaultSyncBatchDuration = 10 * time.Millisecond
 )
 
 var ErrNativeSyncDifferentDomains = errors.New(
@@ -22,15 +21,13 @@ var ErrNativeSyncDifferentDomains = errors.New(
 )
 
 type SyncFactory struct {
-	ReadTimeout   time.Duration
+	ReadTimeout time.Duration
-	BatchDuration time.Duration
+	open        func(string, string, string) (localSyncSource, error)
-	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/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,23 +668,19 @@ func (s *SyncSource) NextSync(ctx context.Context, audioBatch uint64, timeout ti
 			}
 			s.idx++
 
-			var aFrame AudioFrame
+			samples := make([][]byte, s.chans)
-			if aerr == nil {
+			for ch := uint64(0); ch < s.chans; ch++ {
-				samples := make([][]byte, s.chans)
+				f1, f2, _ := av.ChannelFragments(ch)
-				for ch := uint64(0); ch < s.chans; ch++ {
+				if len(f2) > 0 {
-					f1, f2, _ := av.ChannelFragments(ch)
+					samples[ch] = append(f1, f2...)
-					if len(f2) > 0 {
+				} else {
-						samples[ch] = append(f1, f2...)
+					samples[ch] = f1
-					} else {
-						samples[ch] = f1
-					}
-				}
-				aFrame = AudioFrame{
-					Index: aIdx, SampleCount: audioBatch,
-					Channels: s.chans, Samples: samples,
 				}
 			}
-			// even if audio failed, video returns
+			aFrame := AudioFrame{
+				Index: aIdx, SampleCount: audioBatch,
+				Channels: s.chans, Samples: samples,
+			}
 			return vFrame, aFrame, nil
 		case errors.Is(err, mxl.ErrTimeout), errors.Is(err, mxl.ErrOutOfRangeEarly), errors.Is(err, mxl.ErrOutOfRangeLate):
 			timeouts++