add bounded audio source read

.gitignore

@@ -1 +1,2 @@
 build
+imgui.ini

cmd/mxl-player/main.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"io"
 	"log"
+	mxladapter "mxl-player/internal/adapter/mxl"
 	"mxl-player/internal/imgui"
 	"mxl-player/internal/playback"
 	"mxl-player/internal/renderer"
@@ -28,6 +29,17 @@ const (
 	WIN_HEIGHT int32 = 720
 )
 
+const (
+	placeholderWidth  uint32 = 1
+	placeholderHeight uint32 = 1
+	placeholderStride uint32 = 4
+)
+
+const (
+	initialRetryDelay = 500 * time.Millisecond
+	maxRetryDelay     = 5 * time.Second
+)
+
 type appArgs struct {
 	ShowHelp      bool
 	Domain        string
@@ -47,7 +59,7 @@ type appArgs struct {
 
 func printCliHelp(fs *pflag.FlagSet) {
 	fmt.Printf("%s %s\n", APP_NAME, APP_VER)
-	fmt.Println("Usage: mxl-player -d <domain> (-v <uuid> | -a <uuid>) [-f] [-h]")
+	fmt.Println("Usage: mxl-player [-d <domain>] [-v <uuid>] [-a <uuid>] [options]")
 	fmt.Println("                  [-g <gpu-id>] [-p <playback-id>] [--verbose]")
 	fmt.Println("   or: mxl-player [--list-playback] [--list-gpu]")
 	fmt.Println("   or: mxl-player (and set everything in GUI)")
@@ -56,13 +68,20 @@ func printCliHelp(fs *pflag.FlagSet) {
 }
 
 func printUsage(w io.Writer) {
-	fmt.Fprintln(w, "Usage: mxl-player -d <domain> (-v <uuid> | -a <uuid>) [-f] [-h]")
+	fmt.Fprintln(w, "Usage: mxl-player [-d <domain>] [-v <uuid>] [-a <uuid>] [options]")
 	fmt.Fprintln(w, "Try 'mxl-player -h' for more information.")
 }
 
 func checkMXLargs(args appArgs) {
+	if args.VideoFlowId == "" && args.AudioFlowId == "" {
+		return
+	}
+
 	if args.Domain == "" {
-		fmt.Fprintln(os.Stderr, "You should provide valid MXL domain and UUID of at least one flow")
+		fmt.Fprintln(
+			os.Stderr,
+			"You must provide a domain when a feed UUID is configured",
+		)
 		printUsage(os.Stderr)
 		os.Exit(2)
 	}
@@ -74,11 +93,6 @@ func checkMXLargs(args appArgs) {
 		printUsage(os.Stderr)
 		os.Exit(2)
 	}
-	if args.VideoFlowId == "" && args.AudioFlowId == "" {
-		fmt.Fprintln(os.Stderr, "You must provide at least 1 MXL flow UUID")
-		printUsage(os.Stderr)
-		os.Exit(2)
-	}
 }
 
 func main() {
@@ -88,11 +102,18 @@ func main() {
 	// f1 audio: 5fbec3b1-1b0f-417d-9059-8b94a47197eb
 	var args appArgs
 	flagSet := pflag.NewFlagSet(APP_NAME, pflag.ContinueOnError)
+	flagSet.SortFlags = false
 	flagSet.Usage = func() { printUsage(os.Stderr) }
 	flagSet.BoolVarP(&args.ShowHelp, "help", "h", false, "Show help message and exit")
 	flagSet.StringVarP(&args.Domain, "domain", "d", "", "MXL domain directory")
 	flagSet.StringVarP(&args.VideoFlowId, "video", "v", "", "Video flow UUID")
 	flagSet.StringVarP(&args.AudioFlowId, "audio", "a", "", "Audio flow UUID")
+	flagSet.IntVar(
+		&args.MaxAttempts,
+		"max-attempts",
+		0,
+		"Maximum connection attempts per playback lifecycle; 0 retries indefinitely",
+	)
 	flagSet.BoolVarP(&args.IsFullscreen, "fullscreen", "f", false, "Run app in fullscreen mode")
 	flagSet.Uint32VarP(&args.GpuId, "gpu-id", "g", 0, "GPU id [TODO]")
 	flagSet.Uint32VarP(&args.PlaybackId, "playback-id", "p", 0, "Playback audio device id")
@@ -109,10 +130,25 @@ func main() {
 		printCliHelp(flagSet)
 		return
 	}
+	if args.MaxAttempts < 0 {
+		fmt.Fprintln(os.Stderr, "--max-attempts cannot be negative")
+		printUsage(os.Stderr)
+		os.Exit(2)
+	}
+	retryPolicy := playback.RetryPolicy{
+		MaxAttempts:  args.MaxAttempts,
+		InitialDelay: initialRetryDelay,
+		MaxDelay:     maxRetryDelay,
+	}
+	if err := retryPolicy.Validate(); err != nil {
+		fmt.Fprintln(os.Stderr, "invalid retry configuration:", err)
+		os.Exit(2)
+	}
 	if !args.ListAudio && !args.ListGPU {
 		checkMXLargs(args)
 	}
-	// end of cli args parse
+	// path selection
+	useVideoSlot := args.AudioFlowId == ""
 
 	runtime.LockOSThread()
 	if err := sdl.Load(); err != nil {
@@ -265,13 +301,8 @@ func main() {
 		fmt.Printf("sync: video %dx%d audio %dch batch=%d\n",
 			syncSrc.Width(), syncSrc.Height(), aChans, audioBatch)
 	case args.VideoFlowId != "":
-		videoSrc, err = source.Open(args.Domain, args.VideoFlowId)
-		if err != nil {
-			log.Fatalf("source: %v", err)
-		}
-		fmt.Printf("video: %dx%d stride=%d\n", videoSrc.Width(), videoSrc.Height(), videoSrc.Stride())
-
-	default:
+		// VideoSlot owns opening and closing the video reader.
+	case args.AudioFlowId != "":
 		audioSrc, err = source.OpenAudio(args.Domain, args.AudioFlowId)
 		if err != nil {
 			log.Fatalf("audio source: %v", err)
@@ -291,6 +322,8 @@ func main() {
 		}
 		sdl.ResumeAudioStreamDevice(audioStream)
 		fmt.Printf("audio: %dch %d/%d Hz\n", aChans, audioSrc.Rate().Num, audioSrc.Rate().Den)
+	default:
+		// No configured feeds. Renderer and GUI use the placeholder.
 	}
 
 	defer func() {
@@ -308,40 +341,46 @@ func main() {
 		defer sdl.DestroyAudioStream(audioStream)
 	}
 
-	var r *renderer.Renderer
-	if args.VideoFlowId != "" {
-		var w, h, stride uint32
-		if syncSrc != nil {
-			w, h, stride = syncSrc.Width(), syncSrc.Height(), syncSrc.Stride()
-		} else {
-			w, h, stride = videoSrc.Width(), videoSrc.Height(), videoSrc.Stride()
-		}
-		r, err = renderer.New(renderer.Config{
-			PhysDevice:     vkPhysDevice,
-			Device:         vkDevice,
-			Queue:          vkQueue,
-			Surface:        vkSurf,
-			Window:         windowHandler,
-			GraphicsFamily: gfx,
-			VideoWidth:     w,
-			VideoHeight:    h,
-			VideoStride:    stride,
-		})
-		if err != nil {
-			panic(err)
-		}
-		defer r.Destroy()
-		guiBackend, err := imgui.NewVulkanBackend(
-			vkPhysDevice, vkDevice, vkQueue, r.CmdPool(), r.RenderPass())
-		if err != nil {
-			panic(err)
-		}
-		defer guiBackend.Destroy()
-		r.ImGuiDraw = func(cmd vk.CommandBuffer) {
-			guiBackend.RecordDraw(cmd, gui.LastDrawData())
-		}
-		defer vkDevice.WaitIdle()
+	// Create renderer
+	r, err := renderer.New(renderer.Config{
+		PhysDevice:     vkPhysDevice,
+		Device:         vkDevice,
+		Queue:          vkQueue,
+		Surface:        vkSurf,
+		Window:         windowHandler,
+		GraphicsFamily: gfx,
+		VideoWidth:     placeholderWidth,
+		VideoHeight:    placeholderHeight,
+		VideoStride:    placeholderStride,
+	})
+	if err != nil {
+		panic(err)
 	}
+	defer r.Destroy()
+	// Create GUI backend
+	guiBackend, err := imgui.NewVulkanBackend(
+		vkPhysDevice,
+		vkDevice,
+		vkQueue,
+		r.CmdPool(),
+		r.RenderPass(),
+	)
+	if err != nil {
+		panic(err)
+	}
+	defer guiBackend.Destroy()
+	r.ImGuiDraw = func(cmd vk.CommandBuffer) {
+		guiBackend.RecordDraw(cmd, gui.LastDrawData())
+	}
+	if err := r.StageFrame(
+		[]byte{0, 0, 0, 255},
+		placeholderWidth,
+		placeholderHeight,
+		placeholderStride,
+	); err != nil {
+		panic(err)
+	}
+	defer vkDevice.WaitIdle()
 
 	// GUI state (accessible from doReconnect + goroutine)
 	var (
@@ -350,6 +389,7 @@ func main() {
 		audioStr  string = args.AudioFlowId
 		showStats bool   = true
 	)
+	videoActive := args.VideoFlowId != ""
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -358,9 +398,44 @@ func main() {
 		video  string
 		audio  string
 	}
-	// One control channel: grant (empty params) or reconnect (with params).
+	// 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)
+	if err != nil {
+		panic(err)
+	}
+	videoCommands := make(chan playback.FeedConfig, 1)
 
 	reopen := func(params reconnectParams) error {
 		// Close current sources
@@ -377,18 +452,10 @@ func main() {
 			audioSrc = nil
 		}
 		// Try once. Return error if fails — caller loops back to select
-		// and can pick up new reconnect params or a new grant.
+		// 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 {
-				if r != nil {
-					newSize := vk.DeviceSize(s.Stride()) * vk.DeviceSize(s.Height())
-					if newSize != r.FrameSize() {
-						if e = r.RecreateBuffers(newSize); e != nil {
-							return e
-						}
-					}
-				}
 				syncSrc = s
 				aChans = s.Channels()
 				audioBatch = uint64(s.AudioRate().Num) / uint64(s.Rate().Num)
@@ -401,14 +468,6 @@ func main() {
 		} else if params.video != "" {
 			s, e := source.Open(params.domain, params.video)
 			if e == nil {
-				if r != nil {
-					newSize := vk.DeviceSize(s.Stride()) * vk.DeviceSize(s.Height())
-					if newSize != r.FrameSize() {
-						if e = r.RecreateBuffers(newSize); e != nil {
-							return e
-						}
-					}
-				}
 				videoSrc = s
 				return nil
 			}
@@ -429,7 +488,34 @@ func main() {
 		return fmt.Errorf("reopen: no flow specified")
 	}
 
+	enqueueVideoConfig := func(config playback.FeedConfig) {
+		select {
+		case <-videoCommands:
+		default:
+		}
+
+		select {
+		case videoCommands <- config:
+		default:
+		}
+	}
+
 	doReconnect := func() {
+		if useVideoSlot {
+			videoActive = videoStr != ""
+
+			config := playback.FeedConfig{}
+			if videoStr != "" {
+				config = playback.FeedConfig{
+					Domain: domainStr,
+					UUID:   videoStr,
+					Active: true,
+				}
+			}
+
+			enqueueVideoConfig(config)
+			return
+		}
 		select {
 		case <-control:
 		default:
@@ -437,8 +523,27 @@ func main() {
 		control <- reconnectParams{domain: domainStr, video: videoStr, audio: audioStr}
 	}
 
+	playbackDone := make(chan struct{})
 	go func() {
-		// Audio-only mode: independent loop, no grant/staged handshake.
+		defer close(playbackDone)
+
+		if useVideoSlot {
+			err := videoSlot.Run(
+				ctx,
+				playback.FeedConfig{
+					Domain: args.Domain,
+					UUID:   args.VideoFlowId,
+					Active: args.VideoFlowId != "",
+				},
+				videoCommands,
+			)
+			if err != nil && !errors.Is(err, context.Canceled) {
+				log.Printf("video slot: %v", err)
+			}
+			return
+		}
+
+		// Audio-only mode: independent loop.
 		if audioSrc != nil && syncSrc == nil && videoSrc == nil {
 			for {
 				select {
@@ -503,31 +608,34 @@ func main() {
 			}
 		}
 
-		// Video (with or without sync) mode: grant/staged handshake.
+		// Video bridge provides backpressure: only one borrowed frame is in flight.
 		for {
-			params := <-control
-			if params.video != "" || params.audio != "" {
-				// Reconnect request from Connect button or auto-retry.
-				select {
-				case <-control: // drain any pending grant
-				default:
-				}
+			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
@@ -538,15 +646,16 @@ func main() {
 						return
 					}
 					log.Printf("source: %v", err)
-					// Drain any pending grant, then send reconnect.
-					select {
-					case <-control:
-					default:
+					// Request a reconnect after the read failure.
+					params := reconnectParams{
+						domain: domainStr,
+						video:  videoStr,
+						audio:  audioStr,
 					}
 					select {
-					case control <- reconnectParams{domain: domainStr, video: videoStr, audio: audioStr}:
-					case <-ctx.Done():
-						return
+					case control <- params:
+					default:
+						// Preserve an already queued, potentially newer request.
 					}
 					continue
 				}
@@ -570,14 +679,15 @@ func main() {
 						return
 					}
 					log.Printf("source: %v", err)
-					select {
-					case <-control:
-					default:
+					params := reconnectParams{
+						domain: domainStr,
+						video:  videoStr,
+						audio:  audioStr,
 					}
 					select {
-					case control <- reconnectParams{domain: domainStr, video: videoStr, audio: audioStr}:
-					case <-ctx.Done():
-						return
+					case control <- params:
+					default:
+						// Preserve an already queued, potentially newer request.
 					}
 					continue
 				}
@@ -604,12 +714,16 @@ func main() {
 
 	running := true
 	resized := false
-	granted := false
 	fullscreen := args.IsFullscreen
 	if fullscreen {
 		sdl.SetWindowFullscreen(windowHandler, true)
 	}
 	var (
+		displayedVideoWidth  uint32 = placeholderWidth
+		displayedVideoHeight uint32 = placeholderHeight
+		displayedVideoStride uint32 = placeholderStride
+		hasDisplayedVideo    bool   = false
+
 		fps        float64
 		lastIndex  uint64
 		dropped    uint64
@@ -662,15 +776,6 @@ func main() {
 			}
 			resized = false
 		}
-		if !granted {
-			select {
-			case control <- reconnectParams{}:
-				granted = true
-			case <-ctx.Done():
-				running = false
-				continue
-			}
-		}
 		var shownIndex uint64
 		hasFrame := false
 
@@ -697,15 +802,15 @@ func main() {
 			}
 
 			shownIndex = pendingFrame.Frame.Index
-			granted = false
+			displayedVideoWidth = pendingFrame.Frame.Width
+			displayedVideoHeight = pendingFrame.Frame.Height
+			displayedVideoStride = pendingFrame.Frame.Stride
+			hasDisplayedVideo = true
 			hasFrame = true
 		} else if frameErr != nil &&
 			!errors.Is(frameErr, context.DeadlineExceeded) &&
 			!errors.Is(frameErr, context.Canceled) {
 			panic(frameErr)
-		} else {
-			// No frame arrived before the deadline.
-			granted = false
 		}
 
 		// stats
@@ -740,14 +845,12 @@ func main() {
 				cimgui.Text(fmt.Sprintf("FPS: %.1f", fps))
 				cimgui.Text(fmt.Sprintf("Dropped: %d", dropped))
 				cimgui.Text(fmt.Sprintf("Index: %d", shownIndex))
-				if videoSrc != nil {
-					cimgui.Text(fmt.Sprintf("Video: %dx%d %.2fp", videoSrc.Width(), videoSrc.Height(),
-						float32(videoSrc.Rate().Num/videoSrc.Rate().Den)))
-				}
-				if syncSrc != nil {
-					cimgui.Text(fmt.Sprintf("Video: %dx%d %.2fp", syncSrc.Width(), syncSrc.Height(),
-						float32(syncSrc.Rate().Num/syncSrc.Rate().Den)))
-					cimgui.Text(fmt.Sprintf("Audio: %dch %dkHz", syncSrc.Channels(), syncSrc.AudioRate().Num))
+				if hasDisplayedVideo {
+					cimgui.Text(fmt.Sprintf(
+						"Video: %dx%d",
+						displayedVideoWidth,
+						displayedVideoHeight,
+					))
 				}
 				cimgui.Text("\nPress F1 to hide stats")
 				cimgui.Text("Q or Esc to quit")
@@ -762,17 +865,79 @@ func main() {
 			if cimgui.Button("Connect") {
 				doReconnect()
 			}
+			if useVideoSlot && videoActive {
+				cimgui.SameLine()
+				if cimgui.Button("Stop video") {
+					videoActive = false
+					enqueueVideoConfig(
+						playback.FeedConfig{
+							Domain: domainStr,
+							UUID:   videoStr,
+							Active: false,
+						})
+				}
+			}
+			if useVideoSlot && !videoActive && videoStr != "" {
+				cimgui.SameLine()
+				if cimgui.Button("Resume video") {
+					videoActive = true
+					enqueueVideoConfig(playback.FeedConfig{
+						Domain: domainStr,
+						UUID:   videoStr,
+						Active: true,
+					})
+				}
+			}
+			if useVideoSlot && videoStr != "" {
+				if cimgui.Button("Remove video") {
+					videoActive = false
+					videoStr = ""
+					enqueueVideoConfig(playback.FeedConfig{})
+				}
+			}
+			if useVideoSlot {
+				if videoActive {
+					cimgui.Text("Video desired: active")
+				} else if videoStr != "" {
+					cimgui.Text("Video desired: stopped")
+				} else {
+					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")
+				}
+			}
 			cimgui.End()
 			gui.EndFrame()
 			lastFrame = time.Now()
 			// end of test widget
-			var w, h, stride uint32
-			if syncSrc != nil {
-				w, h, stride = syncSrc.Width(), syncSrc.Height(), syncSrc.Stride()
-			} else if videoSrc != nil {
-				w, h, stride = videoSrc.Width(), videoSrc.Height(), videoSrc.Stride()
-			}
-			err := r.DrawFrame(w, h, stride)
+			err := r.DrawFrame(
+				displayedVideoWidth,
+				displayedVideoHeight,
+				displayedVideoStride,
+			)
 			if errors.Is(err, renderer.ErrOutOfDate) {
 				if rerr := r.RecreateSwapchain(); rerr != nil {
 					if errors.Is(rerr, renderer.ErrMinimized) {
@@ -791,4 +956,7 @@ func main() {
 			time.Sleep(10 * time.Millisecond)
 		}
 	}
+
+	cancel()
+	<-playbackDone
 }

imgui.ini

@@ -14,7 +14,7 @@ Size=200,200
 Collapsed=0
 
 [Window][Connection]
-Pos=1250,701
-Size=523,153
+Pos=74,351
+Size=799,226
 Collapsed=0
 

internal/playback/audio.go

@@ -0,0 +1,42 @@
+package playback
+
+import "context"
+
+// AudioFrame contains deinterleaved F32 audio samples.
+//
+// Samples contains one byte slice per channel. Each channel contains
+// SampleCount float32 samples.
+//
+// The sample payload may borrow source-owned memory. AudioSink must finish
+// reading it before ConsumeAudio returns.
+type AudioFrame struct {
+	Index       uint64
+	SampleCount uint64
+	Channels    uint64
+
+	SampleRateNumerator   int64
+	SampleRateDenominator int64
+
+	Samples [][]byte
+}
+
+// AudioReader reads batches from one audio feed.
+//
+// ReadAudio must not be called again until the previous frame has been
+// consumed.
+type AudioReader interface {
+	ReadAudio(context.Context) (AudioFrame, error)
+	Close() error
+}
+
+// AudioReaderFactory opens a reader for the configured audio feed.
+type AudioReaderFactory interface {
+	OpenAudio(context.Context, FeedConfig) (AudioReader, error)
+}
+
+// AudioSink synchronously consumes one borrowed audio batch.
+//
+// ConsumeAudio must not retain frame.Samples or their underlying byte slices.
+type AudioSink interface {
+	ConsumeAudio(context.Context, AudioFrame) error
+}

internal/playback/audio_attempt.go

@@ -0,0 +1,55 @@
+package playback
+
+import (
+	"context"
+	"errors"
+	"fmt"
+)
+
+type audioSinkError struct {
+	err error
+}
+
+func (e *audioSinkError) Error() string {
+	return fmt.Sprintf("consume audio: %v", e.err)
+}
+
+func (e *audioSinkError) Unwrap() error {
+	return e.err
+}
+
+func runAudioAttempt(
+	ctx context.Context,
+	factory AudioReaderFactory,
+	sink AudioSink,
+	config FeedConfig,
+) (resultErr error) {
+	reader, err := factory.OpenAudio(ctx, config)
+	if err != nil {
+		return fmt.Errorf("open audio: %w", err)
+	}
+
+	defer func() {
+		if closeErr := reader.Close(); closeErr != nil {
+			closeErr = fmt.Errorf("close audio: %w", closeErr)
+			resultErr = errors.Join(resultErr, closeErr)
+		}
+	}()
+
+	for {
+		frame, err := reader.ReadAudio(ctx)
+		if err != nil {
+			if ctx.Err() != nil {
+				return ctx.Err()
+			}
+			return fmt.Errorf("read audio: %w", err)
+		}
+
+		if err := sink.ConsumeAudio(ctx, frame); err != nil {
+			if ctx.Err() != nil {
+				return ctx.Err()
+			}
+			return &audioSinkError{err: err}
+		}
+	}
+}

internal/playback/audio_attempt_test.go

@@ -0,0 +1,204 @@
+package playback
+
+import (
+	"context"
+	"errors"
+	"testing"
+)
+
+type fakeAudioFactory struct {
+	reader AudioReader
+	err    error
+	calls  int
+}
+
+func (f *fakeAudioFactory) OpenAudio(
+	context.Context,
+	FeedConfig,
+) (AudioReader, error) {
+	f.calls++
+	return f.reader, f.err
+}
+
+type fakeAudioReader struct {
+	frames    []AudioFrame
+	readErr   error
+	closeErr  error
+	readCalls int
+	closed    bool
+	read      func(context.Context) (AudioFrame, error)
+}
+
+func (r *fakeAudioReader) ReadAudio(ctx context.Context) (AudioFrame, error) {
+	r.readCalls++
+	if r.read != nil {
+		return r.read(ctx)
+	}
+	if len(r.frames) == 0 {
+		return AudioFrame{}, r.readErr
+	}
+	frame := r.frames[0]
+	r.frames = r.frames[1:]
+	return frame, nil
+}
+
+func (r *fakeAudioReader) Close() error {
+	r.closed = true
+	return r.closeErr
+}
+
+type fakeAudioSink struct {
+	frames []AudioFrame
+	err    error
+}
+
+func (s *fakeAudioSink) ConsumeAudio(_ context.Context, frame AudioFrame) error {
+	s.frames = append(s.frames, frame)
+	return s.err
+}
+
+func TestRunAudioAttemptOpenFailure(t *testing.T) {
+	openErr := errors.New("open failed")
+	factory := &fakeAudioFactory{err: openErr}
+	sink := &fakeAudioSink{}
+
+	err := runAudioAttempt(context.Background(), factory, sink, FeedConfig{})
+
+	if !errors.Is(err, openErr) {
+		t.Fatalf("runAudioAttempt() error = %v, want %v", err, openErr)
+	}
+	if factory.calls != 1 {
+		t.Errorf("factory calls = %d, want 1", factory.calls)
+	}
+	if len(sink.frames) != 0 {
+		t.Fatalf("consumed frame count = %d, want 0", len(sink.frames))
+	}
+}
+
+func TestRunAudioAttemptConsumesFrameWithoutCopyThenReturnsReadFailure(t *testing.T) {
+	readErr := errors.New("read failed")
+	wantFrame := AudioFrame{
+		Index:                 42,
+		SampleCount:           2,
+		Channels:              2,
+		SampleRateNumerator:   48000,
+		SampleRateDenominator: 1,
+		Samples: [][]byte{
+			{1, 2, 3, 4},
+			{5, 6, 7, 8},
+		},
+	}
+	reader := &fakeAudioReader{
+		frames:  []AudioFrame{wantFrame},
+		readErr: readErr,
+	}
+	sink := &fakeAudioSink{}
+
+	err := runAudioAttempt(
+		context.Background(),
+		&fakeAudioFactory{reader: reader},
+		sink,
+		FeedConfig{},
+	)
+
+	if !errors.Is(err, readErr) {
+		t.Fatalf("runAudioAttempt() error = %v, want %v", err, readErr)
+	}
+	if !reader.closed {
+		t.Fatal("reader was not closed")
+	}
+	if reader.readCalls != 2 {
+		t.Errorf("read calls = %d, want 2", reader.readCalls)
+	}
+	if len(sink.frames) != 1 {
+		t.Fatalf("consumed frame count = %d, want 1", len(sink.frames))
+	}
+	got := sink.frames[0]
+	if got.Index != wantFrame.Index ||
+		got.SampleCount != wantFrame.SampleCount ||
+		got.Channels != wantFrame.Channels ||
+		got.SampleRateNumerator != wantFrame.SampleRateNumerator ||
+		got.SampleRateDenominator != wantFrame.SampleRateDenominator {
+		t.Errorf("consumed frame metadata = %+v, want %+v", got, wantFrame)
+	}
+	for channel := range wantFrame.Samples {
+		if &got.Samples[channel][0] != &wantFrame.Samples[channel][0] {
+			t.Fatalf("channel %d samples were copied", channel)
+		}
+	}
+}
+
+func TestRunAudioAttemptSinkFailureStopsReadingAndCloses(t *testing.T) {
+	sinkErr := errors.New("audio output unavailable")
+	reader := &fakeAudioReader{
+		frames: []AudioFrame{
+			{Index: 1, Samples: [][]byte{{1}}},
+			{Index: 2, Samples: [][]byte{{2}}},
+		},
+	}
+
+	err := runAudioAttempt(
+		context.Background(),
+		&fakeAudioFactory{reader: reader},
+		&fakeAudioSink{err: sinkErr},
+		FeedConfig{},
+	)
+
+	if !errors.Is(err, sinkErr) {
+		t.Fatalf("runAudioAttempt() error = %v, want %v", err, sinkErr)
+	}
+	var typedErr *audioSinkError
+	if !errors.As(err, &typedErr) {
+		t.Fatalf("runAudioAttempt() error type = %T, want *audioSinkError", err)
+	}
+	if reader.readCalls != 1 {
+		t.Errorf("read calls = %d, want 1", reader.readCalls)
+	}
+	if !reader.closed {
+		t.Fatal("reader was not closed")
+	}
+}
+
+func TestRunAudioAttemptCanceledRead(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	reader := &fakeAudioReader{
+		read: func(ctx context.Context) (AudioFrame, error) {
+			cancel()
+			return AudioFrame{}, ctx.Err()
+		},
+	}
+
+	err := runAudioAttempt(
+		ctx,
+		&fakeAudioFactory{reader: reader},
+		&fakeAudioSink{},
+		FeedConfig{},
+	)
+
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("runAudioAttempt() error = %v, want context.Canceled", err)
+	}
+	if !reader.closed {
+		t.Fatal("reader was not closed")
+	}
+}
+
+func TestRunAudioAttemptJoinsReadAndCloseErrors(t *testing.T) {
+	readErr := errors.New("read failed")
+	closeErr := errors.New("close failed")
+	reader := &fakeAudioReader{readErr: readErr, closeErr: closeErr}
+
+	err := runAudioAttempt(
+		context.Background(),
+		&fakeAudioFactory{reader: reader},
+		&fakeAudioSink{},
+		FeedConfig{},
+	)
+
+	if !errors.Is(err, readErr) {
+		t.Errorf("runAudioAttempt() error does not contain read error: %v", err)
+	}
+	if !errors.Is(err, closeErr) {
+		t.Errorf("runAudioAttempt() error does not contain close error: %v", err)
+	}
+}

internal/playback/audio_slot.go

@@ -0,0 +1,93 @@
+package playback
+
+import (
+	"context"
+	"errors"
+	"fmt"
+)
+
+var ErrAudioWorkerRequired = errors.New("audio worker is required")
+
+type AudioSlot struct {
+	worker *AudioWorker
+}
+
+func NewAudioSlot(worker *AudioWorker) (*AudioSlot, error) {
+	if worker == nil {
+		return nil, ErrAudioWorkerRequired
+	}
+	return &AudioSlot{worker: worker}, nil
+}
+
+func (s *AudioSlot) Run(
+	ctx context.Context,
+	initial FeedConfig,
+	commands <-chan FeedConfig,
+) error {
+	if err := initial.Validate(); err != nil {
+		return fmt.Errorf("validate initial audio config: %w", err)
+	}
+
+	var (
+		workerCancel context.CancelFunc
+		workerDone   chan error
+	)
+
+	start := func(config FeedConfig) {
+		workerCtx, cancel := context.WithCancel(ctx)
+		done := make(chan error, 1)
+
+		workerCancel = cancel
+		workerDone = done
+
+		go func() {
+			done <- s.worker.Run(workerCtx, config)
+		}()
+	}
+
+	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/audio_slot_test.go

@@ -0,0 +1,202 @@
+package playback
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+)
+
+type slotAudioFactory struct {
+	opened chan FeedConfig
+
+	mu         sync.Mutex
+	active     int
+	maxActive  int
+	closeCount int
+}
+
+func newSlotAudioFactory() *slotAudioFactory {
+	return &slotAudioFactory{opened: make(chan FeedConfig, 8)}
+}
+
+func (f *slotAudioFactory) OpenAudio(
+	_ context.Context,
+	config FeedConfig,
+) (AudioReader, error) {
+	f.mu.Lock()
+	f.active++
+	if f.active > f.maxActive {
+		f.maxActive = f.active
+	}
+	f.mu.Unlock()
+	f.opened <- config
+	return &slotAudioReader{factory: f}, nil
+}
+
+func (f *slotAudioFactory) counts() (active, maxActive, closeCount int) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.active, f.maxActive, f.closeCount
+}
+
+type slotAudioReader struct {
+	factory *slotAudioFactory
+}
+
+func (r *slotAudioReader) ReadAudio(ctx context.Context) (AudioFrame, error) {
+	<-ctx.Done()
+	return AudioFrame{}, ctx.Err()
+}
+
+func (r *slotAudioReader) Close() error {
+	r.factory.mu.Lock()
+	defer r.factory.mu.Unlock()
+	r.factory.active--
+	r.factory.closeCount++
+	return nil
+}
+
+func newAudioSlotWorker(t *testing.T, factory AudioReaderFactory) *AudioWorker {
+	t.Helper()
+	return newAudioWorkerForTest(
+		t,
+		factory,
+		&fakeAudioSink{},
+		1,
+		func(error) bool { return false },
+		nil,
+	)
+}
+
+func receiveAudioSlotOpen(t *testing.T, opened <-chan FeedConfig) FeedConfig {
+	t.Helper()
+	select {
+	case config := <-opened:
+		return config
+	case <-time.After(time.Second):
+		t.Fatal("audio worker did not open")
+		return FeedConfig{}
+	}
+}
+
+func TestNewAudioSlotRequiresWorker(t *testing.T) {
+	slot, err := NewAudioSlot(nil)
+	if slot != nil {
+		t.Fatalf("NewAudioSlot(nil) slot = %#v, want nil", slot)
+	}
+	if !errors.Is(err, ErrAudioWorkerRequired) {
+		t.Fatalf("NewAudioSlot(nil) error = %v, want %v", err, ErrAudioWorkerRequired)
+	}
+}
+
+func TestAudioSlotStartsAndJoinsInitialWorker(t *testing.T) {
+	factory := newSlotAudioFactory()
+	slot, err := NewAudioSlot(newAudioSlotWorker(t, factory))
+	if err != nil {
+		t.Fatalf("NewAudioSlot() error = %v", err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	want := FeedConfig{Domain: "/audio", UUID: "first", Active: true}
+
+	go func() { done <- slot.Run(ctx, want, make(chan FeedConfig)) }()
+	if got := receiveAudioSlotOpen(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 %v", err, context.Canceled)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("Run() did not stop after cancellation")
+	}
+	active, maxActive, closeCount := factory.counts()
+	if active != 0 || maxActive != 1 || closeCount != 1 {
+		t.Fatalf("reader counts = %d, %d, %d; want 0, 1, 1", active, maxActive, closeCount)
+	}
+}
+
+func TestAudioSlotReplacesWithoutOverlapAndStops(t *testing.T) {
+	factory := newSlotAudioFactory()
+	slot, err := NewAudioSlot(newAudioSlotWorker(t, factory))
+	if err != nil {
+		t.Fatalf("NewAudioSlot() error = %v", err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	commands := make(chan FeedConfig)
+	done := make(chan error, 1)
+	first := FeedConfig{Domain: "/audio", UUID: "first", Active: true}
+	second := FeedConfig{Domain: "/audio", UUID: "second", Active: true}
+
+	go func() { done <- slot.Run(ctx, first, commands) }()
+	receiveAudioSlotOpen(t, factory.opened)
+	commands <- second
+	if got := receiveAudioSlotOpen(t, factory.opened); got != second {
+		t.Fatalf("replacement config = %#v, want %#v", got, second)
+	}
+	commands <- FeedConfig{Domain: "/audio", UUID: "second", Active: false}
+
+	deadline := time.Now().Add(time.Second)
+	for {
+		active, maxActive, closeCount := factory.counts()
+		if active == 0 && closeCount == 2 {
+			if maxActive != 1 {
+				t.Fatalf("maximum active readers = %d, want 1", maxActive)
+			}
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("reader counts = %d, %d, %d; want 0, 1, 2", active, maxActive, closeCount)
+		}
+		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")
+	}
+}
+
+func TestAudioSlotIgnoresInvalidCommand(t *testing.T) {
+	factory := newSlotAudioFactory()
+	slot, err := NewAudioSlot(newAudioSlotWorker(t, factory))
+	if err != nil {
+		t.Fatalf("NewAudioSlot() error = %v", err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	commands := make(chan FeedConfig)
+	done := make(chan error, 1)
+
+	go func() {
+		done <- slot.Run(
+			ctx,
+			FeedConfig{Domain: "/audio", UUID: "first", Active: true},
+			commands,
+		)
+	}()
+	receiveAudioSlotOpen(t, factory.opened)
+	commands <- FeedConfig{UUID: "invalid", Active: true}
+
+	select {
+	case config := <-factory.opened:
+		t.Fatalf("invalid command opened config %#v", config)
+	case <-time.After(20 * time.Millisecond):
+	}
+	active, _, closeCount := factory.counts()
+	if active != 1 || closeCount != 0 {
+		t.Fatalf("invalid command disturbed reader: active %d, closed %d", active, closeCount)
+	}
+	cancel()
+	<-done
+}

internal/playback/audio_worker.go

@@ -0,0 +1,184 @@
+package playback
+
+import (
+	"context"
+	"errors"
+	"fmt"
+)
+
+var (
+	ErrAudioFactoryRequired      = errors.New("audio reader factory is required")
+	ErrAudioSinkRequired         = errors.New("audio sink is required")
+	ErrAudioRetryDeciderRequired = errors.New("audio decider is required")
+	ErrAudioFeedInactive         = errors.New("audio feed is not active")
+)
+
+type AudioWorker struct {
+	factory     AudioReaderFactory
+	sink        AudioSink
+	retry       RetryPolicy
+	shouldRetry retryDecider
+	observer    StatusObserver
+	wait        waitFunc
+}
+
+func NewAudioWorker(
+	factory AudioReaderFactory,
+	sink AudioSink,
+	retry RetryPolicy,
+	shouldRetry func(error) bool,
+	observer StatusObserver,
+) (*AudioWorker, error) {
+	if factory == nil {
+		return nil, ErrAudioFactoryRequired
+	}
+	if sink == nil {
+		return nil, ErrAudioSinkRequired
+	}
+	if shouldRetry == nil {
+		return nil, ErrAudioRetryDeciderRequired
+	}
+	if err := retry.Validate(); err != nil {
+		return nil, fmt.Errorf("validate audio retry policy: %w", err)
+	}
+
+	return &AudioWorker{
+		factory:     factory,
+		sink:        sink,
+		retry:       retry,
+		shouldRetry: shouldRetry,
+		observer:    observer,
+		wait:        waitForRetry,
+	}, nil
+}
+
+type stabilityAudioSink struct {
+	sink     AudioSink
+	onStable func()
+	stable   bool
+}
+
+func (s *stabilityAudioSink) ConsumeAudio(
+	ctx context.Context,
+	frame AudioFrame,
+) error {
+	err := s.sink.ConsumeAudio(ctx, frame)
+	if err == nil && !s.stable {
+		s.stable = true
+		if s.onStable != nil {
+			s.onStable()
+		}
+	}
+	return err
+}
+
+func (w *AudioWorker) emit(status Status) {
+	if w.observer != nil {
+		w.observer(status)
+	}
+}
+
+func (w *AudioWorker) Run(
+	ctx context.Context,
+	config FeedConfig,
+) error {
+	if err := config.Validate(); err != nil {
+		return fmt.Errorf("validate audio config: %w", err)
+	}
+	if !config.Active {
+		return ErrAudioFeedInactive
+	}
+
+	attemptNumber := 0
+	var latestRetry retryEvent
+
+	attempt := func(ctx context.Context) (bool, error) {
+		attemptNumber++
+
+		state := StateConnecting
+		if attemptNumber > 1 {
+			state = StateReconnecting
+		}
+		w.emit(Status{
+			Unit:    UnitAudio,
+			State:   state,
+			Attempt: attemptNumber,
+		})
+
+		attemptSink := &stabilityAudioSink{
+			sink: w.sink,
+			onStable: func() {
+				w.emit(Status{
+					Unit:    UnitAudio,
+					State:   StatePlaying,
+					Attempt: attemptNumber,
+				})
+			},
+		}
+
+		err := runAudioAttempt(ctx, w.factory, attemptSink, config)
+		return attemptSink.stable, err
+	}
+
+	decide := func(err error) bool {
+		var sinkErr *audioSinkError
+		if errors.As(err, &sinkErr) {
+			return false
+		}
+		return w.shouldRetry(err)
+	}
+
+	observeRetry := func(event retryEvent) {
+		latestRetry = event
+		if !event.WillRetry {
+			return
+		}
+
+		w.emit(Status{
+			Unit:           UnitAudio,
+			State:          StateReconnecting,
+			Attempt:        attemptNumber + 1,
+			FailedAttempts: event.FailedAttempts,
+			RetryIn:        event.RetryIn,
+			Err:            event.Err,
+		})
+	}
+
+	err := runWithRetry(
+		ctx,
+		w.retry,
+		attempt,
+		decide,
+		w.wait,
+		observeRetry,
+	)
+
+	if ctx.Err() != nil {
+		w.emit(Status{
+			Unit:  UnitAudio,
+			State: StateStopping,
+		})
+		w.emit(Status{
+			Unit:  UnitAudio,
+			State: StateIdle,
+		})
+		return ctx.Err()
+	}
+
+	if err != nil {
+		w.emit(Status{
+			Unit:           UnitAudio,
+			State:          StateFailed,
+			Attempt:        attemptNumber,
+			FailedAttempts: latestRetry.FailedAttempts,
+			Err:            err,
+		})
+		return err
+	}
+
+	w.emit(Status{
+		Unit:  UnitAudio,
+		State: StateIdle,
+	})
+	return nil
+}

internal/playback/audio_worker_test.go

@@ -0,0 +1,241 @@
+package playback
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+)
+
+type queuedAudioFactory struct {
+	readers []AudioReader
+	errs    []error
+	calls   int
+}
+
+func (f *queuedAudioFactory) OpenAudio(
+	context.Context,
+	FeedConfig,
+) (AudioReader, error) {
+	index := f.calls
+	f.calls++
+	if index < len(f.errs) && f.errs[index] != nil {
+		return nil, f.errs[index]
+	}
+	if index < len(f.readers) {
+		return f.readers[index], nil
+	}
+	return nil, errors.New("unexpected audio open")
+}
+
+func newAudioWorkerForTest(
+	t *testing.T,
+	factory AudioReaderFactory,
+	sink AudioSink,
+	maxAttempts int,
+	shouldRetry func(error) bool,
+	observer StatusObserver,
+) *AudioWorker {
+	t.Helper()
+	worker, err := NewAudioWorker(
+		factory,
+		sink,
+		testRetryPolicy(maxAttempts),
+		shouldRetry,
+		observer,
+	)
+	if err != nil {
+		t.Fatalf("NewAudioWorker() error = %v", err)
+	}
+	worker.wait = func(context.Context, time.Duration) error { return nil }
+	return worker
+}
+
+func TestNewAudioWorkerValidatesDependencies(t *testing.T) {
+	factory := &fakeAudioFactory{}
+	sink := &fakeAudioSink{}
+	retry := testRetryPolicy(3)
+	decider := func(error) bool { return true }
+
+	tests := []struct {
+		name    string
+		factory AudioReaderFactory
+		sink    AudioSink
+		retry   RetryPolicy
+		decider func(error) bool
+		wantErr error
+	}{
+		{name: "factory", sink: sink, retry: retry, decider: decider, wantErr: ErrAudioFactoryRequired},
+		{name: "sink", factory: factory, retry: retry, decider: decider, wantErr: ErrAudioSinkRequired},
+		{name: "decider", factory: factory, sink: sink, retry: retry, wantErr: ErrAudioRetryDeciderRequired},
+		{
+			name:    "retry policy",
+			factory: factory,
+			sink:    sink,
+			retry:   RetryPolicy{},
+			decider: decider,
+			wantErr: ErrInvalidRetryDelay,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			worker, err := NewAudioWorker(
+				tt.factory,
+				tt.sink,
+				tt.retry,
+				tt.decider,
+				nil,
+			)
+			if worker != nil {
+				t.Fatalf("NewAudioWorker() worker = %#v, want nil", worker)
+			}
+			if !errors.Is(err, tt.wantErr) {
+				t.Fatalf("NewAudioWorker() error = %v, want %v", err, tt.wantErr)
+			}
+		})
+	}
+}
+
+func TestAudioWorkerRejectsInactiveFeed(t *testing.T) {
+	worker := newAudioWorkerForTest(
+		t,
+		&fakeAudioFactory{},
+		&fakeAudioSink{},
+		1,
+		func(error) bool { return false },
+		nil,
+	)
+
+	err := worker.Run(
+		context.Background(),
+		FeedConfig{Domain: "/audio", UUID: "audio", Active: false},
+	)
+	if !errors.Is(err, ErrAudioFeedInactive) {
+		t.Fatalf("Run() error = %v, want %v", err, ErrAudioFeedInactive)
+	}
+}
+
+func TestAudioWorkerPublishesPlayingThenFailed(t *testing.T) {
+	readErr := errors.New("audio disappeared")
+	reader := &fakeAudioReader{
+		frames:  []AudioFrame{{Index: 1, Samples: [][]byte{{1, 2, 3, 4}}}},
+		readErr: readErr,
+	}
+	var statuses []Status
+	worker := newAudioWorkerForTest(
+		t,
+		&fakeAudioFactory{reader: reader},
+		&fakeAudioSink{},
+		1,
+		func(error) bool { return false },
+		func(status Status) { statuses = append(statuses, status) },
+	)
+
+	err := worker.Run(
+		context.Background(),
+		FeedConfig{Domain: "/audio", UUID: "audio", Active: true},
+	)
+	if !errors.Is(err, readErr) {
+		t.Fatalf("Run() error = %v, want %v", err, readErr)
+	}
+	wantStates := []State{StateConnecting, StatePlaying, StateFailed}
+	if len(statuses) != len(wantStates) {
+		t.Fatalf("status count = %d, want %d: %#v", len(statuses), len(wantStates), statuses)
+	}
+	for i, want := range wantStates {
+		if statuses[i].Unit != UnitAudio || statuses[i].State != want {
+			t.Errorf("status[%d] = %#v, want audio/%v", i, statuses[i], want)
+		}
+	}
+}
+
+func TestAudioWorkerRetriesUpToAttemptLimit(t *testing.T) {
+	openErr := errors.New("audio unavailable")
+	factory := &queuedAudioFactory{errs: []error{openErr, openErr, openErr}}
+	var statuses []Status
+	worker := newAudioWorkerForTest(
+		t,
+		factory,
+		&fakeAudioSink{},
+		3,
+		func(error) bool { return true },
+		func(status Status) { statuses = append(statuses, status) },
+	)
+
+	err := worker.Run(
+		context.Background(),
+		FeedConfig{Domain: "/audio", UUID: "audio", Active: true},
+	)
+	if !errors.Is(err, openErr) {
+		t.Fatalf("Run() error = %v, want %v", err, openErr)
+	}
+	if factory.calls != 3 {
+		t.Fatalf("open calls = %d, want 3", factory.calls)
+	}
+	last := statuses[len(statuses)-1]
+	if last.State != StateFailed || last.Attempt != 3 || last.FailedAttempts != 3 {
+		t.Fatalf("last status = %#v, want failed attempt 3", last)
+	}
+}
+
+func TestAudioWorkerDoesNotRetrySinkFailure(t *testing.T) {
+	sinkErr := errors.New("output failed")
+	factory := &fakeAudioFactory{
+		reader: &fakeAudioReader{frames: []AudioFrame{{Index: 1}}},
+	}
+	worker := newAudioWorkerForTest(
+		t,
+		factory,
+		&fakeAudioSink{err: sinkErr},
+		3,
+		func(error) bool { return true },
+		nil,
+	)
+
+	err := worker.Run(
+		context.Background(),
+		FeedConfig{Domain: "/audio", UUID: "audio", Active: true},
+	)
+	if !errors.Is(err, sinkErr) {
+		t.Fatalf("Run() error = %v, want %v", err, sinkErr)
+	}
+	if factory.calls != 1 {
+		t.Fatalf("open calls = %d, want 1", factory.calls)
+	}
+}
+
+func TestAudioWorkerCancellationPublishesStoppingAndIdle(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	reader := &fakeAudioReader{
+		read: func(ctx context.Context) (AudioFrame, error) {
+			cancel()
+			<-ctx.Done()
+			return AudioFrame{}, ctx.Err()
+		},
+	}
+	var statuses []Status
+	worker := newAudioWorkerForTest(
+		t,
+		&fakeAudioFactory{reader: reader},
+		&fakeAudioSink{},
+		1,
+		func(error) bool { return true },
+		func(status Status) { statuses = append(statuses, status) },
+	)
+
+	err := worker.Run(
+		ctx,
+		FeedConfig{Domain: "/audio", UUID: "audio", Active: true},
+	)
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("Run() error = %v, want %v", err, context.Canceled)
+	}
+	if len(statuses) < 3 {
+		t.Fatalf("statuses = %#v, want connecting, stopping, idle", statuses)
+	}
+	last := statuses[len(statuses)-2:]
+	if last[0].State != StateStopping || last[1].State != StateIdle {
+		t.Fatalf("final statuses = %#v, want stopping then idle", last)
+	}
+}

internal/playback/state.go

@@ -1,6 +1,9 @@
 package playback
 
-import "time"
+import (
+	"fmt"
+	"time"
+)
 
 type Unit uint8
 
@@ -31,3 +34,35 @@ type Status struct {
 }
 
 type StatusObserver func(Status)
+
+func (u Unit) String() string {
+	switch u {
+	case UnitVideo:
+		return "video"
+	case UnitAudio:
+		return "audio"
+	case UnitSync:
+		return "sync"
+	default:
+		return fmt.Sprintf("Unit(%d)", uint8(u))
+	}
+}
+
+func (s State) String() string {
+	switch s {
+	case StateIdle:
+		return "idle"
+	case StateConnecting:
+		return "connecting"
+	case StatePlaying:
+		return "playing"
+	case StateReconnecting:
+		return "reconnecting"
+	case StateFailed:
+		return "failed"
+	case StateStopping:
+		return "stopping"
+	default:
+		return fmt.Sprintf("State(%d)", uint8(s))
+	}
+}

internal/playback/state_test.go

@@ -38,3 +38,42 @@ func TestStatusPreservesValues(t *testing.T) {
 		t.Errorf("RetryIn = %s, want %s", status.RetryIn, time.Second)
 	}
 }
+
+func TestUnitString(t *testing.T) {
+	tests := []struct {
+		unit Unit
+		want string
+	}{
+		{unit: UnitVideo, want: "video"},
+		{unit: UnitAudio, want: "audio"},
+		{unit: UnitSync, want: "sync"},
+		{unit: Unit(255), want: "Unit(255)"},
+	}
+
+	for _, tt := range tests {
+		if got := tt.unit.String(); got != tt.want {
+			t.Errorf("Unit(%d).String() = %q, want %q", tt.unit, got, tt.want)
+		}
+	}
+}
+
+func TestStateString(t *testing.T) {
+	tests := []struct {
+		state State
+		want  string
+	}{
+		{state: StateIdle, want: "idle"},
+		{state: StateConnecting, want: "connecting"},
+		{state: StatePlaying, want: "playing"},
+		{state: StateReconnecting, want: "reconnecting"},
+		{state: StateFailed, want: "failed"},
+		{state: StateStopping, want: "stopping"},
+		{state: State(255), want: "State(255)"},
+	}
+
+	for _, tt := range tests {
+		if got := tt.state.String(); got != tt.want {
+			t.Errorf("State(%d).String() = %q, want %q", tt.state, got, tt.want)
+		}
+	}
+}

internal/playback/status_store.go

@@ -0,0 +1,33 @@
+package playback
+
+import "sync"
+
+type StatusStore struct {
+	mu       sync.RWMutex
+	statuses map[Unit]Status
+}
+
+func NewStatusStore() *StatusStore {
+	return &StatusStore{
+		statuses: make(map[Unit]Status),
+	}
+}
+
+func (s *StatusStore) Observe(status Status) {
+	s.mu.Lock()
+	s.statuses[status.Unit] = status
+	s.mu.Unlock()
+}
+
+func (s *StatusStore) Snapshot(unit Unit) (Status, bool) {
+	s.mu.RLock()
+	status, ok := s.statuses[unit]
+	s.mu.RUnlock()
+	return status, ok
+}
+
+func (s *StatusStore) Clear(unit Unit) {
+	s.mu.Lock()
+	delete(s.statuses, unit)
+	s.mu.Unlock()
+}

internal/playback/status_store_test.go

@@ -0,0 +1,108 @@
+package playback
+
+import (
+	"errors"
+	"sync"
+	"testing"
+)
+
+func TestStatusStoreSnapshotUnknownUnit(t *testing.T) {
+	store := NewStatusStore()
+
+	status, ok := store.Snapshot(UnitVideo)
+	if ok {
+		t.Fatalf("Snapshot() = %#v, true; want false", status)
+	}
+}
+
+func TestStatusStoreKeepsUnitsIndependent(t *testing.T) {
+	store := NewStatusStore()
+	wantVideo := Status{
+		Unit:           UnitVideo,
+		State:          StateReconnecting,
+		Attempt:        3,
+		FailedAttempts: 2,
+		Err:            errors.New("video unavailable"),
+	}
+	wantAudio := Status{
+		Unit:    UnitAudio,
+		State:   StatePlaying,
+		Attempt: 1,
+	}
+
+	store.Observe(wantVideo)
+	store.Observe(wantAudio)
+
+	if got, ok := store.Snapshot(UnitVideo); !ok || got != wantVideo {
+		t.Fatalf("video Snapshot() = %#v, %t; want %#v, true", got, ok, wantVideo)
+	}
+	if got, ok := store.Snapshot(UnitAudio); !ok || got != wantAudio {
+		t.Fatalf("audio Snapshot() = %#v, %t; want %#v, true", got, ok, wantAudio)
+	}
+}
+
+func TestStatusStoreObserveReplacesLatestStatus(t *testing.T) {
+	store := NewStatusStore()
+	store.Observe(Status{Unit: UnitVideo, State: StateConnecting, Attempt: 1})
+	want := Status{Unit: UnitVideo, State: StatePlaying, Attempt: 2}
+	store.Observe(want)
+
+	got, ok := store.Snapshot(UnitVideo)
+	if !ok || got != want {
+		t.Fatalf("Snapshot() = %#v, %t; want %#v, true", got, ok, want)
+	}
+}
+
+func TestStatusStoreClearOnlySelectedUnit(t *testing.T) {
+	store := NewStatusStore()
+	wantAudio := Status{Unit: UnitAudio, State: StatePlaying}
+	store.Observe(Status{Unit: UnitVideo, State: StatePlaying})
+	store.Observe(wantAudio)
+
+	store.Clear(UnitVideo)
+
+	if status, ok := store.Snapshot(UnitVideo); ok {
+		t.Fatalf("video Snapshot() = %#v, true after Clear", status)
+	}
+	if got, ok := store.Snapshot(UnitAudio); !ok || got != wantAudio {
+		t.Fatalf("audio Snapshot() = %#v, %t; want %#v, true", got, ok, wantAudio)
+	}
+}
+
+func TestStatusStoreConcurrentAccess(t *testing.T) {
+	store := NewStatusStore()
+	const iterations = 1000
+
+	var writers sync.WaitGroup
+	for _, unit := range []Unit{UnitVideo, UnitAudio, UnitSync} {
+		unit := unit
+		writers.Add(1)
+		go func() {
+			defer writers.Done()
+			for attempt := 1; attempt <= iterations; attempt++ {
+				store.Observe(Status{
+					Unit:    unit,
+					State:   StatePlaying,
+					Attempt: attempt,
+				})
+				store.Snapshot(unit)
+			}
+		}()
+	}
+	writers.Wait()
+
+	for _, unit := range []Unit{UnitVideo, UnitAudio, UnitSync} {
+		status, ok := store.Snapshot(unit)
+		if !ok {
+			t.Fatalf("Snapshot(%v) not found", unit)
+		}
+		if status.Attempt != iterations {
+			t.Fatalf(
+				"Snapshot(%v) attempt = %d, want %d",
+				unit,
+				status.Attempt,
+				iterations,
+			)
+		}
+	}
+}

internal/playback/video_slot.go

@@ -0,0 +1,92 @@
+package playback
+
+import (
+	"context"
+	"errors"
+	"fmt"
+)
+
+var ErrVideoWorkerRequired = errors.New("video worker is required")
+
+type VideoSlot struct {
+	worker *VideoWorker
+}
+
+func NewVideoSlot(worker *VideoWorker) (*VideoSlot, error) {
+	if worker == nil {
+		return nil, ErrVideoWorkerRequired
+	}
+	return &VideoSlot{worker: worker}, nil
+}
+
+func (s *VideoSlot) Run(
+	ctx context.Context,
+	initial FeedConfig,
+	commands <-chan FeedConfig,
+) error {
+	if err := initial.Validate(); err != nil {
+		return fmt.Errorf("validate initial video config: %w", err)
+	}
+
+	var (
+		workerCancel context.CancelFunc
+		workerDone   chan error
+	)
+
+	start := func(config FeedConfig) {
+		workerCtx, cancel := context.WithCancel(ctx)
+		done := make(chan error, 1)
+
+		workerCancel = cancel
+		workerDone = done
+
+		go func() {
+			done <- s.worker.Run(workerCtx, config)
+		}()
+	}
+
+	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.
+			// Clear its lifecycle, but keep the slot alive for future commands.
+			workerCancel()
+			workerCancel = nil
+			workerDone = nil
+		}
+	}
+}

internal/playback/video_slot_test.go

@@ -0,0 +1,233 @@
+package playback
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+)
+
+type slotVideoFactory struct {
+	opened chan FeedConfig
+
+	mu         sync.Mutex
+	active     int
+	maxActive  int
+	closeCount int
+}
+
+func newSlotVideoFactory() *slotVideoFactory {
+	return &slotVideoFactory{opened: make(chan FeedConfig, 8)}
+}
+
+func (f *slotVideoFactory) OpenVideo(
+	_ context.Context,
+	config FeedConfig,
+) (VideoReader, error) {
+	f.mu.Lock()
+	f.active++
+	if f.active > f.maxActive {
+		f.maxActive = f.active
+	}
+	f.mu.Unlock()
+
+	f.opened <- config
+	return &slotVideoReader{factory: f}, nil
+}
+
+func (f *slotVideoFactory) counts() (active, maxActive, closeCount int) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.active, f.maxActive, f.closeCount
+}
+
+type slotVideoReader struct {
+	factory *slotVideoFactory
+}
+
+func (r *slotVideoReader) ReadVideo(ctx context.Context) (VideoFrame, error) {
+	<-ctx.Done()
+	return VideoFrame{}, ctx.Err()
+}
+
+func (r *slotVideoReader) Close() error {
+	r.factory.mu.Lock()
+	defer r.factory.mu.Unlock()
+	r.factory.active--
+	r.factory.closeCount++
+	return nil
+}
+
+func newSlotTestWorker(t *testing.T, factory VideoReaderFactory) *VideoWorker {
+	t.Helper()
+	worker, err := NewVideoWorker(
+		factory,
+		&fakeVideoSink{},
+		testRetryPolicy(1),
+		func(error) bool { return false },
+		nil,
+	)
+	if err != nil {
+		t.Fatalf("NewVideoWorker() error = %v", err)
+	}
+	return worker
+}
+
+func receiveSlotOpen(t *testing.T, opened <-chan FeedConfig) FeedConfig {
+	t.Helper()
+	select {
+	case config := <-opened:
+		return config
+	case <-time.After(time.Second):
+		t.Fatal("video worker did not open")
+		return FeedConfig{}
+	}
+}
+
+func TestNewVideoSlotRequiresWorker(t *testing.T) {
+	slot, err := NewVideoSlot(nil)
+	if slot != nil {
+		t.Fatalf("NewVideoSlot(nil) slot = %#v, want nil", slot)
+	}
+	if !errors.Is(err, ErrVideoWorkerRequired) {
+		t.Fatalf("NewVideoSlot(nil) error = %v, want %v", err, ErrVideoWorkerRequired)
+	}
+}
+
+func TestVideoSlotStartsInitialActiveConfig(t *testing.T) {
+	factory := newSlotVideoFactory()
+	slot, err := NewVideoSlot(newSlotTestWorker(t, factory))
+	if err != nil {
+		t.Fatalf("NewVideoSlot() error = %v", err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	want := FeedConfig{Domain: "/video", UUID: "first", Active: true}
+
+	go func() { done <- slot.Run(ctx, want, make(chan FeedConfig)) }()
+	if got := receiveSlotOpen(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 %v", err, context.Canceled)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("Run() did not stop after cancellation")
+	}
+	active, _, closeCount := factory.counts()
+	if active != 0 || closeCount != 1 {
+		t.Fatalf("reader counts = active %d, closed %d; want 0, 1", active, closeCount)
+	}
+}
+
+func TestVideoSlotReplacesWithoutOverlappingWorkers(t *testing.T) {
+	factory := newSlotVideoFactory()
+	slot, err := NewVideoSlot(newSlotTestWorker(t, factory))
+	if err != nil {
+		t.Fatalf("NewVideoSlot() error = %v", err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	commands := make(chan FeedConfig)
+	done := make(chan error, 1)
+	first := FeedConfig{Domain: "/video", UUID: "first", Active: true}
+	second := FeedConfig{Domain: "/video", UUID: "second", Active: true}
+
+	go func() { done <- slot.Run(ctx, first, commands) }()
+	receiveSlotOpen(t, factory.opened)
+	commands <- second
+	if got := receiveSlotOpen(t, factory.opened); got != second {
+		t.Fatalf("replacement config = %#v, want %#v", got, second)
+	}
+
+	cancel()
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("Run() did not stop")
+	}
+	active, maxActive, closeCount := factory.counts()
+	if active != 0 || maxActive != 1 || closeCount != 2 {
+		t.Fatalf(
+			"reader counts = active %d, maximum %d, closed %d; want 0, 1, 2",
+			active, maxActive, closeCount,
+		)
+	}
+}
+
+func TestVideoSlotIgnoresInvalidCommand(t *testing.T) {
+	factory := newSlotVideoFactory()
+	slot, err := NewVideoSlot(newSlotTestWorker(t, factory))
+	if err != nil {
+		t.Fatalf("NewVideoSlot() error = %v", err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	commands := make(chan FeedConfig)
+	done := make(chan error, 1)
+	initial := FeedConfig{Domain: "/video", UUID: "first", Active: true}
+
+	go func() { done <- slot.Run(ctx, initial, commands) }()
+	receiveSlotOpen(t, factory.opened)
+	commands <- FeedConfig{UUID: "invalid", Active: true}
+
+	select {
+	case config := <-factory.opened:
+		t.Fatalf("invalid command opened config %#v", config)
+	case <-time.After(20 * time.Millisecond):
+	}
+
+	active, _, closeCount := factory.counts()
+	if active != 1 || closeCount != 0 {
+		t.Fatalf("invalid command disturbed reader: active %d, closed %d", active, closeCount)
+	}
+	cancel()
+	<-done
+}
+
+func TestVideoSlotInactiveCommandStopsWithoutRestart(t *testing.T) {
+	factory := newSlotVideoFactory()
+	slot, err := NewVideoSlot(newSlotTestWorker(t, factory))
+	if err != nil {
+		t.Fatalf("NewVideoSlot() error = %v", err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	commands := make(chan FeedConfig)
+	done := make(chan error, 1)
+	initial := FeedConfig{Domain: "/video", UUID: "first", Active: true}
+
+	go func() { done <- slot.Run(ctx, initial, commands) }()
+	receiveSlotOpen(t, factory.opened)
+	commands <- FeedConfig{Domain: "/video", UUID: "first", Active: false}
+
+	deadline := time.Now().Add(time.Second)
+	for {
+		active, _, closeCount := factory.counts()
+		if active == 0 && closeCount == 1 {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatal("inactive command did not stop reader")
+		}
+		time.Sleep(time.Millisecond)
+	}
+	select {
+	case config := <-factory.opened:
+		t.Fatalf("inactive command restarted config %#v", config)
+	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 after commands closed")
+	}
+}

internal/source/source.go

@@ -296,47 +296,98 @@ func OpenAudio(domain, flowID string) (*AudioSource, error) {
 	}, nil
 }
 
-func (s *AudioSource) NextAudio(ctx context.Context, batch uint64, timeout time.Duration) (AudioFrame, error) {
-	for {
-		select {
-		case <-ctx.Done():
-			return AudioFrame{}, ctx.Err()
-		default:
+func (s *AudioSource) ReadAudioOnceCtx(
+	ctx context.Context,
+	batch uint64,
+	timeout time.Duration,
+) (AudioFrame, error) {
+	if err := ctx.Err(); err != nil {
+		return AudioFrame{}, err
+	}
+
+	value, err := s.r.GetSamples(s.idx, int(batch), timeout)
+	if ctxErr := ctx.Err(); ctxErr != nil {
+		return AudioFrame{}, ctxErr
+	}
+
+	switch {
+	case err == nil:
+		samples := make([][]byte, s.chans)
+		for channel := uint64(0); channel < s.chans; channel++ {
+			first, second, _ := value.ChannelFragments(channel)
+			if len(second) > 0 {
+				samples[channel] = append(first, second...)
+			} else {
+				samples[channel] = first
+			}
 		}
-		v, err := s.r.GetSamples(s.idx, int(batch), timeout)
-		switch {
-		case err == nil:
-			samples := make([][]byte, s.chans)
-			for ch := uint64(0); ch < s.chans; ch++ {
-				f1, f2, _ := v.ChannelFragments(ch)
-				if len(f2) > 0 {
-					samples[ch] = append(f1, f2...)
-				} else {
-					samples[ch] = f1
+
+		frame := AudioFrame{
+			Index:       s.idx,
+			SampleCount: batch,
+			Channels:    s.chans,
+			Samples:     samples,
+		}
+		s.idx += batch
+		return frame, nil
+	case errors.Is(err, mxl.ErrOutOfRangeEarly):
+		return AudioFrame{}, wrapError(
+			"read audio",
+			ErrorKindTemporary,
+			err,
+		)
+	case errors.Is(err, mxl.ErrOutOfRangeLate):
+		runtimeInfo, runtimeErr := s.r.Runtime()
+		if runtimeErr != nil {
+			return AudioFrame{}, wrapError(
+				"read audio runtime",
+				ErrorKindUnavailable,
+				runtimeErr,
+			)
+		}
+		s.idx = runtimeInfo.HeadIndex
+		return AudioFrame{}, wrapError(
+			"read audio",
+			ErrorKindTemporary,
+			err,
+		)
+	default:
+		return AudioFrame{}, wrapError(
+			"read audio",
+			ErrorKindUnavailable,
+			err,
+		)
+	}
+}
+
+func (s *AudioSource) NextAudio(
+	ctx context.Context,
+	batch uint64,
+	timeout time.Duration,
+) (AudioFrame, error) {
+	for {
+		frame, err := s.ReadAudioOnceCtx(ctx, batch, timeout)
+		if err == nil {
+			return frame, nil
+		}
+		if ctx.Err() != nil {
+			return AudioFrame{}, ctx.Err()
+		}
+		if KindOf(err) != ErrorKindTemporary {
+			return AudioFrame{}, err
+		}
+
+		timer := time.NewTimer(10 * time.Millisecond)
+		select {
+		case <-timer.C:
+		case <-ctx.Done():
+			if !timer.Stop() {
+				select {
+				case <-timer.C:
+				default:
 				}
 			}
-			f := AudioFrame{
-				Index:       s.idx,
-				SampleCount: batch,
-				Channels:    s.chans,
-				Samples:     samples,
-			}
-			s.idx += batch
-			return f, nil
-		case errors.Is(err, mxl.ErrOutOfRangeEarly):
-			select {
-			case <-time.After(10 * time.Millisecond):
-			case <-ctx.Done():
-				return AudioFrame{}, ctx.Err()
-			}
-		case errors.Is(err, mxl.ErrOutOfRangeLate):
-			rt, rerr := s.r.Runtime()
-			if rerr != nil {
-				return AudioFrame{}, fmt.Errorf("Runtime: %w", rerr)
-			}
-			s.idx = rt.HeadIndex
-		default:
-			return AudioFrame{}, fmt.Errorf("GetSamples: %w", err)
+			return AudioFrame{}, ctx.Err()
 		}
 	}
 }

notes.md

@@ -0,0 +1,6 @@
+That makes sense. The GUI is likely more responsive because frame staging is now serialized with rendering:
+- The background goroutine no longer copies a large frame into Vulkan-mapped memory concurrently with GUI/render work.
+- StageFrame waits for the GPU fence before writing, removing CPU/GPU memory contention and undefined synchronization.
+- The bridge creates deterministic backpressure: the source cannot begin another frame until the current payload is staged.
+- The main thread now controls the complete render sequence instead of coordinating loosely through two channels.
+So we fixed both correctness and scheduling stability without adding another frame copy.