stage borrowed video frames on render thread

cmd/mxl-player/main.go

@@ -7,6 +7,7 @@ import (
 	"io"
 	"log"
 	"mxl-player/internal/imgui"
+	"mxl-player/internal/playback"
 	"mxl-player/internal/renderer"
 	"mxl-player/internal/sdl"
 	"mxl-player/internal/source"
@@ -359,7 +360,7 @@ func main() {
 	}
 	// One control channel: grant (empty params) or reconnect (with params).
 	control := make(chan reconnectParams, 1)
-	staged := make(chan uint64)
+	videoBridge := playback.NewVideoBridge()
 
 	reopen := func(params reconnectParams) error {
 		// Close current sources
@@ -529,9 +530,7 @@ func main() {
 				continue
 			}
 
-			var payload []byte
+			var videoFrame playback.VideoFrame
-			var grainIdx uint64
-
 			if syncSrc != nil {
 				vFrame, aFrame, err := syncSrc.NextSync(ctx, audioBatch, 200*time.Millisecond)
 				if err != nil {
@@ -551,8 +550,15 @@ func main() {
 					}
 					continue
 				}
-				payload = vFrame.Payload
+				videoFrame = playback.VideoFrame{
-				grainIdx = vFrame.Index
+					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))
 				}
@@ -575,16 +581,22 @@ func main() {
 					}
 					continue
 				}
-				payload = f.Payload
+				videoFrame = playback.VideoFrame{
-				grainIdx = f.Index
+					Index:   f.Index,
+					Width:   f.Width,
+					Height:  f.Height,
+					Stride:  f.Stride,
+					Size:    f.Size,
+					Invalid: f.Invalid,
+					Payload: f.Payload,
+				}
 			}
 
-			if r != nil {
+			if err := videoBridge.ConsumeVideo(ctx, videoFrame); err != nil {
-				vk.CopyToMapped(r.StagingMapped(), payload)
+				if errors.Is(err, context.Canceled) {
+					return
 				}
-			select {
+				log.Printf("video output: %v", err)
-			case staged <- grainIdx:
-			case <-ctx.Done():
 				return
 			}
 		}
@@ -661,15 +673,38 @@ func main() {
 		}
 		var shownIndex uint64
 		hasFrame := false
-		select {
+
-		case shownIndex = <-staged:
+		frameCtx, frameCancel := context.WithTimeout(ctx, 100*time.Millisecond)
+		pendingFrame, frameErr := videoBridge.Next(frameCtx)
+		frameCancel()
+
+		if pendingFrame != nil {
+			var stageErr error
+			if r != nil {
+				stageErr = r.StageFrame(
+					pendingFrame.Frame.Payload,
+					pendingFrame.Frame.Width,
+					pendingFrame.Frame.Height,
+					pendingFrame.Frame.Stride,
+				)
+			}
+
+			// Release the borrowed payload before reacting to a staging error
+			pendingFrame.Complete(stageErr)
+
+			if stageErr != nil {
+				panic(stageErr)
+			}
+
+			shownIndex = pendingFrame.Frame.Index
 			granted = false
 			hasFrame = true
-		case <-ctx.Done():
+		} else if frameErr != nil &&
-			running = false
+			!errors.Is(frameErr, context.DeadlineExceeded) &&
-			continue
+			!errors.Is(frameErr, context.Canceled) {
-		case <-time.After(100 * time.Millisecond):
+			panic(frameErr)
-			// No frame staged. Reset granted so we re-grant on next iteration.
+		} else {
+			// No frame arrived before the deadline.
 			granted = false
 		}
 

imgui.ini

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

internal/adapter/mxl/video.go

@@ -10,15 +10,43 @@ import (
 	"mxl-player/internal/source"
 )
 
-const DefaultVideoReadTimeout = 200 * time.Millisecond
+const (
+	DefaultVideoReadTimeout      = 200 * time.Millisecond
+	DefaultVideoUnavailableAfter = 2 * time.Second
+	DefaultTemporaryRetryDelay   = 10 * time.Millisecond
+)
 
 type VideoFactory struct {
 	ReadTimeout      time.Duration
+	UnavailableAfter time.Duration
 }
 
 type videoReader struct {
-	source  *source.Source
+	source           localVideoSource
-	timeout time.Duration
+	readTimeout      time.Duration
+	unavailableAfter time.Duration
+	retryDelay       time.Duration
+	now              func() time.Time
+	wait             temporaryWaitFunc
+}
+
+type localVideoSource interface {
+	ReadOnceCtx(context.Context, time.Duration) (source.Frame, error)
+	Close() error
+}
+
+type temporaryWaitFunc func(context.Context, time.Duration) error
+
+func waitForTemporaryRetry(ctx context.Context, delay time.Duration) error {
+	timer := time.NewTimer(delay)
+	defer timer.Stop()
+
+	select {
+	case <-timer.C:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
 }
 
 var _ playback.VideoReaderFactory = VideoFactory{}
@@ -56,25 +84,34 @@ func (f VideoFactory) OpenVideo(
 		return nil, err
 	}
 
-	timeout := f.ReadTimeout
+	readTimeout := f.ReadTimeout
-	if timeout <= 0 {
+	if readTimeout <= 0 {
-		timeout = DefaultVideoReadTimeout
+		readTimeout = DefaultVideoReadTimeout
+	}
+
+	unavailableAfter := f.UnavailableAfter
+	if unavailableAfter <= 0 {
+		unavailableAfter = DefaultVideoUnavailableAfter
 	}
 
 	return &videoReader{
 		source:           src,
-		timeout: timeout,
+		readTimeout:      readTimeout,
+		unavailableAfter: unavailableAfter,
+		retryDelay:       DefaultTemporaryRetryDelay,
+		now:              time.Now,
+		wait:             waitForTemporaryRetry,
 	}, nil
 }
 
 func (r *videoReader) ReadVideo(
 	ctx context.Context,
 ) (playback.VideoFrame, error) {
-	frame, err := r.source.NextCtx(ctx, r.timeout)
+	var unavailableSince time.Time
-	if err != nil {
-		return playback.VideoFrame{}, err
-	}
 
+	for {
+		frame, err := r.source.ReadOnceCtx(ctx, r.readTimeout)
+		if err == nil {
 			return playback.VideoFrame{
 				Index:   frame.Index,
 				Width:   frame.Width,
@@ -84,6 +121,36 @@ func (r *videoReader) ReadVideo(
 				Invalid: frame.Invalid,
 				Payload: frame.Payload,
 			}, nil
+		}
+		if ctx.Err() != nil {
+			return playback.VideoFrame{}, ctx.Err()
+		}
+		if source.KindOf(err) != source.ErrorKindTemporary {
+			return playback.VideoFrame{}, err
+		}
+
+		now := r.now()
+		if unavailableSince.IsZero() {
+			unavailableSince = now
+		} else if now.Sub(unavailableSince) >= r.unavailableAfter {
+			return playback.VideoFrame{}, &source.SourceError{
+				Op:   "read local MXL video",
+				Kind: source.ErrorKindUnavailable,
+				Err: fmt.Errorf(
+					"no video data for %s: %w",
+					r.unavailableAfter,
+					err,
+				),
+			}
+		}
+
+		if err := r.wait(ctx, r.retryDelay); err != nil {
+			if ctx.Err() != nil {
+				return playback.VideoFrame{}, ctx.Err()
+			}
+			return playback.VideoFrame{}, err
+		}
+	}
 }
 
 func (r *videoReader) Close() error {

internal/adapter/mxl/video_integration_test.go

@@ -0,0 +1,95 @@
+package mxladapter
+
+import (
+	"context"
+	"errors"
+	"os"
+	"testing"
+	"time"
+
+	"mxl-player/internal/playback"
+)
+
+type cancelingVideoSink struct {
+	cancel      context.CancelFunc
+	width       uint32
+	height      uint32
+	payloadSize int
+	got         bool
+}
+
+func (s *cancelingVideoSink) ConsumeVideo(
+	_ context.Context,
+	frame playback.VideoFrame,
+) error {
+	s.width = frame.Width
+	s.height = frame.Height
+	s.payloadSize = len(frame.Payload)
+	s.got = true
+	s.cancel()
+	return nil
+}
+
+func TestVideoWorkerIntegration(t *testing.T) {
+	domain := os.Getenv("MXL_TEST_VIDEO_DOMAIN")
+	uuid := os.Getenv("MXL_TEST_VIDEO_UUID")
+	if domain == "" || uuid == "" {
+		t.Skip("set MXL_TEST_VIDEO_DOMAIN and MXL_TEST_VIDEO_UUID")
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+	defer cancel()
+
+	sink := &cancelingVideoSink{cancel: cancel}
+	var statuses []playback.Status
+
+	worker, err := playback.NewVideoWorker(
+		VideoFactory{},
+		sink,
+		playback.RetryPolicy{
+			MaxAttempts:  1,
+			InitialDelay: 100 * time.Millisecond,
+			MaxDelay:     time.Second,
+		},
+		ShouldRetry,
+		func(status playback.Status) {
+			statuses = append(statuses, status)
+		},
+	)
+	if err != nil {
+		t.Fatalf("NewVideoWorker() error = %v", err)
+	}
+
+	err = worker.Run(ctx, playback.FeedConfig{
+		Domain: domain,
+		UUID:   uuid,
+		Active: true,
+	})
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("Run() error = %v, want context.Canceled", err)
+	}
+	if !sink.got {
+		t.Fatal("worker did not deliver a video frame")
+	}
+	if sink.width == 0 || sink.height == 0 {
+		t.Fatalf(
+			"invalid frame dimensions: %dx%d",
+			sink.width,
+			sink.height,
+		)
+	}
+	if sink.payloadSize == 0 {
+		t.Fatal("video frame payload is empty")
+	}
+
+	foundPlaying := false
+	for _, status := range statuses {
+		if status.State == playback.StatePlaying {
+			foundPlaying = true
+			break
+		}
+	}
+	if !foundPlaying {
+		t.Fatalf("statuses contain no Playing transition: %+v", statuses)
+	}
+}

internal/adapter/mxl/video_test.go

@@ -4,11 +4,51 @@ import (
 	"context"
 	"errors"
 	"testing"
+	"time"
 
 	"mxl-player/internal/playback"
 	"mxl-player/internal/source"
 )
 
+type localVideoReadResult struct {
+	frame source.Frame
+	err   error
+}
+
+type fakeLocalVideoSource struct {
+	results  []localVideoReadResult
+	calls    int
+	timeouts []time.Duration
+	closeErr error
+	closed   bool
+}
+
+func (s *fakeLocalVideoSource) ReadOnceCtx(
+	_ context.Context,
+	timeout time.Duration,
+) (source.Frame, error) {
+	s.timeouts = append(s.timeouts, timeout)
+	if s.calls >= len(s.results) {
+		return source.Frame{}, errors.New("unexpected local video read")
+	}
+	result := s.results[s.calls]
+	s.calls++
+	return result.frame, result.err
+}
+
+func (s *fakeLocalVideoSource) Close() error {
+	s.closed = true
+	return s.closeErr
+}
+
+func temporaryVideoError(cause error) error {
+	return &source.SourceError{
+		Op:   "read video",
+		Kind: source.ErrorKindTemporary,
+		Err:  cause,
+	}
+}
+
 func TestVideoFactoryOpenVideoCanceled(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	cancel()
@@ -73,3 +113,188 @@ func TestVideoFactoryOpenVideoRejectsInvalidConfig(t *testing.T) {
 		})
 	}
 }
+
+func TestVideoReaderTemporaryFailureThenFrame(t *testing.T) {
+	temporaryErr := errors.New("video is early")
+	payload := []byte{1, 2, 3, 4}
+	wantFrame := source.Frame{
+		Index:   42,
+		Width:   1920,
+		Height:  1080,
+		Stride:  5120,
+		Size:    5120 * 1080,
+		Invalid: false,
+		Payload: payload,
+	}
+	localSource := &fakeLocalVideoSource{
+		results: []localVideoReadResult{
+			{err: temporaryVideoError(temporaryErr)},
+			{frame: wantFrame},
+		},
+	}
+	waits := 0
+	reader := &videoReader{
+		source:           localSource,
+		readTimeout:      250 * time.Millisecond,
+		unavailableAfter: 2 * time.Second,
+		retryDelay:       10 * time.Millisecond,
+		now:              func() time.Time { return time.Unix(100, 0) },
+		wait: func(context.Context, time.Duration) error {
+			waits++
+			return nil
+		},
+	}
+
+	got, err := reader.ReadVideo(context.Background())
+	if err != nil {
+		t.Fatalf("ReadVideo() error = %v, want nil", err)
+	}
+	if localSource.calls != 2 {
+		t.Errorf("source read calls = %d, want 2", localSource.calls)
+	}
+	if waits != 1 {
+		t.Errorf("temporary wait calls = %d, want 1", waits)
+	}
+	if len(localSource.timeouts) != 2 ||
+		localSource.timeouts[0] != 250*time.Millisecond ||
+		localSource.timeouts[1] != 250*time.Millisecond {
+		t.Errorf("source timeouts = %v, want [250ms 250ms]", localSource.timeouts)
+	}
+	if got.Index != wantFrame.Index ||
+		got.Width != wantFrame.Width ||
+		got.Height != wantFrame.Height ||
+		got.Stride != wantFrame.Stride ||
+		got.Size != wantFrame.Size ||
+		got.Invalid != wantFrame.Invalid {
+		t.Errorf("video frame = %+v, want metadata from %+v", got, wantFrame)
+	}
+	if len(got.Payload) == 0 || &got.Payload[0] != &payload[0] {
+		t.Fatal("video payload was copied")
+	}
+}
+
+func TestVideoReaderProlongedTemporaryFailureBecomesUnavailable(t *testing.T) {
+	lastCause := errors.New("video timeout")
+	localSource := &fakeLocalVideoSource{
+		results: []localVideoReadResult{
+			{err: temporaryVideoError(errors.New("first timeout"))},
+			{err: temporaryVideoError(errors.New("second timeout"))},
+			{err: temporaryVideoError(lastCause)},
+		},
+	}
+	times := []time.Time{
+		time.Unix(100, 0),
+		time.Unix(101, 0),
+		time.Unix(102, 0),
+	}
+	nowCall := 0
+	waits := 0
+	reader := &videoReader{
+		source:           localSource,
+		readTimeout:      200 * time.Millisecond,
+		unavailableAfter: 2 * time.Second,
+		retryDelay:       10 * time.Millisecond,
+		now: func() time.Time {
+			result := times[nowCall]
+			nowCall++
+			return result
+		},
+		wait: func(context.Context, time.Duration) error {
+			waits++
+			return nil
+		},
+	}
+
+	_, err := reader.ReadVideo(context.Background())
+	if err == nil {
+		t.Fatal("ReadVideo() error is nil after prolonged unavailability")
+	}
+	if got := source.KindOf(err); got != source.ErrorKindUnavailable {
+		t.Fatalf("source.KindOf(ReadVideo()) = %v, want %v", got, source.ErrorKindUnavailable)
+	}
+	if !errors.Is(err, lastCause) {
+		t.Errorf("ReadVideo() error = %v, want cause %v", err, lastCause)
+	}
+	if !ShouldRetry(err) {
+		t.Error("ShouldRetry(ReadVideo()) = false, want true")
+	}
+	if localSource.calls != 3 {
+		t.Errorf("source read calls = %d, want 3", localSource.calls)
+	}
+	if waits != 2 {
+		t.Errorf("temporary wait calls = %d, want 2", waits)
+	}
+}
+
+func TestVideoReaderReturnsNonTemporaryErrorImmediately(t *testing.T) {
+	unavailableErr := &source.SourceError{
+		Op:   "read video",
+		Kind: source.ErrorKindUnavailable,
+		Err:  errors.New("flow invalid"),
+	}
+	localSource := &fakeLocalVideoSource{
+		results: []localVideoReadResult{{err: unavailableErr}},
+	}
+	reader := &videoReader{
+		source:           localSource,
+		readTimeout:      200 * time.Millisecond,
+		unavailableAfter: 2 * time.Second,
+		retryDelay:       10 * time.Millisecond,
+		now: func() time.Time {
+			t.Fatal("clock called for non-temporary error")
+			return time.Time{}
+		},
+		wait: func(context.Context, time.Duration) error {
+			t.Fatal("wait called for non-temporary error")
+			return nil
+		},
+	}
+
+	_, err := reader.ReadVideo(context.Background())
+	if !errors.Is(err, unavailableErr) {
+		t.Fatalf("ReadVideo() error = %v, want %v", err, unavailableErr)
+	}
+	if localSource.calls != 1 {
+		t.Errorf("source read calls = %d, want 1", localSource.calls)
+	}
+}
+
+func TestVideoReaderCancellationDuringTemporaryWait(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	localSource := &fakeLocalVideoSource{
+		results: []localVideoReadResult{{err: temporaryVideoError(errors.New("early"))}},
+	}
+	reader := &videoReader{
+		source:           localSource,
+		readTimeout:      200 * time.Millisecond,
+		unavailableAfter: 2 * time.Second,
+		retryDelay:       10 * time.Millisecond,
+		now:              func() time.Time { return time.Unix(100, 0) },
+		wait: func(ctx context.Context, _ time.Duration) error {
+			cancel()
+			return ctx.Err()
+		},
+	}
+
+	_, err := reader.ReadVideo(ctx)
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("ReadVideo() error = %v, want context.Canceled", err)
+	}
+	if localSource.calls != 1 {
+		t.Errorf("source read calls = %d, want 1", localSource.calls)
+	}
+}
+
+func TestVideoReaderCloseDelegatesToSource(t *testing.T) {
+	closeErr := errors.New("close failed")
+	localSource := &fakeLocalVideoSource{closeErr: closeErr}
+	reader := &videoReader{source: localSource}
+
+	err := reader.Close()
+	if !errors.Is(err, closeErr) {
+		t.Fatalf("Close() error = %v, want %v", err, closeErr)
+	}
+	if !localSource.closed {
+		t.Fatal("local source was not closed")
+	}
+}

internal/playback/video_bridge.go

@@ -0,0 +1,60 @@
+package playback
+
+import (
+	"context"
+	"sync"
+)
+
+type PendingVideoFrame struct {
+	Frame VideoFrame
+
+	completeOnce sync.Once
+	result       chan error
+}
+
+type VideoBridge struct {
+	requests chan *PendingVideoFrame
+}
+
+func NewVideoBridge() *VideoBridge {
+	return &VideoBridge{
+		requests: make(chan *PendingVideoFrame),
+	}
+}
+
+func (b *VideoBridge) ConsumeVideo(
+	ctx context.Context,
+	frame VideoFrame,
+) error {
+	pending := &PendingVideoFrame{
+		Frame:  frame,
+		result: make(chan error, 1),
+	}
+
+	select {
+	case b.requests <- pending:
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+
+	// The render thread now owns temporary access to the borrowed payload.
+	// We must wait for Complete even if ctx is canceled.
+	return <-pending.result
+}
+
+func (b *VideoBridge) Next(
+	ctx context.Context,
+) (*PendingVideoFrame, error) {
+	select {
+	case pending := <-b.requests:
+		return pending, nil
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+}
+
+func (f *PendingVideoFrame) Complete(err error) {
+	f.completeOnce.Do(func() {
+		f.result <- err
+	})
+}

internal/playback/video_bridge_test.go

@@ -0,0 +1,138 @@
+package playback
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+)
+
+const videoBridgeTestTimeout = time.Second
+
+func TestVideoBridgeDeliversFrameAndCompletionResult(t *testing.T) {
+	bridge := NewVideoBridge()
+	wantErr := errors.New("stage frame")
+	wantFrame := VideoFrame{
+		Index:   42,
+		Width:   1920,
+		Height:  1080,
+		Stride:  7680,
+		Payload: []byte{1, 2, 3},
+	}
+	consumeResult := make(chan error, 1)
+
+	go func() {
+		consumeResult <- bridge.ConsumeVideo(context.Background(), wantFrame)
+	}()
+
+	ctx, cancel := context.WithTimeout(context.Background(), videoBridgeTestTimeout)
+	defer cancel()
+	pending, err := bridge.Next(ctx)
+	if err != nil {
+		t.Fatalf("Next() error = %v", err)
+	}
+	if pending.Frame.Index != wantFrame.Index {
+		t.Fatalf("Next() frame index = %d, want %d", pending.Frame.Index, wantFrame.Index)
+	}
+	if &pending.Frame.Payload[0] != &wantFrame.Payload[0] {
+		t.Fatal("Next() copied the borrowed payload")
+	}
+
+	pending.Complete(wantErr)
+	select {
+	case err := <-consumeResult:
+		if !errors.Is(err, wantErr) {
+			t.Fatalf("ConsumeVideo() error = %v, want %v", err, wantErr)
+		}
+	case <-time.After(videoBridgeTestTimeout):
+		t.Fatal("ConsumeVideo() did not return after completion")
+	}
+}
+
+func TestVideoBridgeConsumeHonorsCancellationBeforeDelivery(t *testing.T) {
+	bridge := NewVideoBridge()
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	err := bridge.ConsumeVideo(ctx, VideoFrame{})
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("ConsumeVideo() error = %v, want %v", err, context.Canceled)
+	}
+}
+
+func TestVideoBridgeConsumeWaitsForCompletionAfterDelivery(t *testing.T) {
+	bridge := NewVideoBridge()
+	ctx, cancel := context.WithCancel(context.Background())
+	consumeResult := make(chan error, 1)
+
+	go func() {
+		consumeResult <- bridge.ConsumeVideo(ctx, VideoFrame{Index: 7})
+	}()
+
+	nextCtx, nextCancel := context.WithTimeout(context.Background(), videoBridgeTestTimeout)
+	defer nextCancel()
+	pending, err := bridge.Next(nextCtx)
+	if err != nil {
+		t.Fatalf("Next() error = %v", err)
+	}
+	cancel()
+
+	select {
+	case err := <-consumeResult:
+		t.Fatalf("ConsumeVideo() returned before completion: %v", err)
+	case <-time.After(20 * time.Millisecond):
+	}
+
+	pending.Complete(nil)
+	select {
+	case err := <-consumeResult:
+		if err != nil {
+			t.Fatalf("ConsumeVideo() error = %v, want nil", err)
+		}
+	case <-time.After(videoBridgeTestTimeout):
+		t.Fatal("ConsumeVideo() did not return after completion")
+	}
+}
+
+func TestPendingVideoFrameCompleteIsIdempotent(t *testing.T) {
+	bridge := NewVideoBridge()
+	consumeResult := make(chan error, 1)
+
+	go func() {
+		consumeResult <- bridge.ConsumeVideo(context.Background(), VideoFrame{})
+	}()
+
+	ctx, cancel := context.WithTimeout(context.Background(), videoBridgeTestTimeout)
+	defer cancel()
+	pending, err := bridge.Next(ctx)
+	if err != nil {
+		t.Fatalf("Next() error = %v", err)
+	}
+
+	firstErr := errors.New("first")
+	pending.Complete(firstErr)
+	pending.Complete(errors.New("second"))
+
+	select {
+	case err := <-consumeResult:
+		if !errors.Is(err, firstErr) {
+			t.Fatalf("ConsumeVideo() error = %v, want %v", err, firstErr)
+		}
+	case <-time.After(videoBridgeTestTimeout):
+		t.Fatal("ConsumeVideo() did not return")
+	}
+}
+
+func TestVideoBridgeNextHonorsCancellation(t *testing.T) {
+	bridge := NewVideoBridge()
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	pending, err := bridge.Next(ctx)
+	if pending != nil {
+		t.Fatalf("Next() pending = %#v, want nil", pending)
+	}
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("Next() error = %v, want %v", err, context.Canceled)
+	}
+}

internal/playback/video_worker.go

@@ -0,0 +1,184 @@
+package playback
+
+import (
+	"context"
+	"errors"
+	"fmt"
+)
+
+var (
+	ErrVideoFactoryRequired      = errors.New("video reader factory is required")
+	ErrVideoSinkRequired         = errors.New("video sink is required")
+	ErrVideoRetryDeciderRequired = errors.New("video decider is required")
+	ErrVideoFeedInactive         = errors.New("video feed is not active")
+)
+
+type VideoWorker struct {
+	factory     VideoReaderFactory
+	sink        VideoSink
+	retry       RetryPolicy
+	shouldRetry retryDecider
+	observer    StatusObserver
+	wait        waitFunc
+}
+
+func NewVideoWorker(
+	factory VideoReaderFactory,
+	sink VideoSink,
+	retry RetryPolicy,
+	shouldRetry func(error) bool,
+	observer StatusObserver,
+) (*VideoWorker, error) {
+	if factory == nil {
+		return nil, ErrVideoFactoryRequired
+	}
+	if sink == nil {
+		return nil, ErrVideoSinkRequired
+	}
+	if shouldRetry == nil {
+		return nil, ErrVideoRetryDeciderRequired
+	}
+	if err := retry.Validate(); err != nil {
+		return nil, fmt.Errorf("validate video retry policy: %w", err)
+	}
+
+	return &VideoWorker{
+		factory:     factory,
+		sink:        sink,
+		retry:       retry,
+		shouldRetry: shouldRetry,
+		observer:    observer,
+		wait:        waitForRetry,
+	}, nil
+}
+
+type stabilityVideoSink struct {
+	sink     VideoSink
+	onStable func()
+	stable   bool
+}
+
+func (s *stabilityVideoSink) ConsumeVideo(
+	ctx context.Context,
+	frame VideoFrame,
+) error {
+	err := s.sink.ConsumeVideo(ctx, frame)
+	if err == nil && !s.stable {
+		s.stable = true
+		if s.onStable != nil {
+			s.onStable()
+		}
+	}
+	return err
+}
+
+func (w *VideoWorker) emit(status Status) {
+	if w.observer != nil {
+		w.observer(status)
+	}
+}
+
+func (w *VideoWorker) Run(
+	ctx context.Context,
+	config FeedConfig,
+) error {
+	if err := config.Validate(); err != nil {
+		return fmt.Errorf("validate video config: %w", err)
+	}
+	if !config.Active {
+		return ErrVideoFeedInactive
+	}
+
+	attemptNumber := 0
+	var latestRetry retryEvent
+
+	attempt := func(ctx context.Context) (bool, error) {
+		attemptNumber++
+
+		state := StateConnecting
+		if attemptNumber > 1 {
+			state = StateReconnecting
+		}
+		w.emit(Status{
+			Unit:    UnitVideo,
+			State:   state,
+			Attempt: attemptNumber,
+		})
+
+		attemptSink := &stabilityVideoSink{
+			sink: w.sink,
+			onStable: func() {
+				w.emit(Status{
+					Unit:    UnitVideo,
+					State:   StatePlaying,
+					Attempt: attemptNumber,
+				})
+			},
+		}
+
+		err := runVideoAttempt(ctx, w.factory, attemptSink, config)
+		return attemptSink.stable, err
+	}
+
+	decide := func(err error) bool {
+		var sinkErr *videoSinkError
+		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:           UnitVideo,
+			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:  UnitVideo,
+			State: StateStopping,
+		})
+		w.emit(Status{
+			Unit:  UnitVideo,
+			State: StateIdle,
+		})
+		return ctx.Err()
+	}
+
+	if err != nil {
+		w.emit(Status{
+			Unit:           UnitVideo,
+			State:          StateFailed,
+			Attempt:        attemptNumber,
+			FailedAttempts: latestRetry.FailedAttempts,
+			Err:            err,
+		})
+		return err
+	}
+
+	w.emit(Status{
+		Unit:  UnitVideo,
+		State: StateIdle,
+	})
+	return nil
+}

internal/playback/video_worker_test.go

@@ -0,0 +1,319 @@
+package playback
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+)
+
+type videoOpenResult struct {
+	reader VideoReader
+	err    error
+}
+
+type scriptedVideoFactory struct {
+	results []videoOpenResult
+	calls   int
+}
+
+func (f *scriptedVideoFactory) OpenVideo(
+	context.Context,
+	FeedConfig,
+) (VideoReader, error) {
+	if f.calls >= len(f.results) {
+		return nil, errors.New("unexpected video open attempt")
+	}
+	result := f.results[f.calls]
+	f.calls++
+	return result.reader, result.err
+}
+
+func activeVideoConfig() FeedConfig {
+	return FeedConfig{
+		Domain: "/dev/shm/mxl",
+		UUID:   "video-uuid",
+		Active: true,
+	}
+}
+
+func newTestVideoWorker(
+	t *testing.T,
+	factory VideoReaderFactory,
+	sink VideoSink,
+	maxAttempts int,
+	shouldRetry func(error) bool,
+	observer StatusObserver,
+) *VideoWorker {
+	t.Helper()
+
+	worker, err := NewVideoWorker(
+		factory,
+		sink,
+		testRetryPolicy(maxAttempts),
+		shouldRetry,
+		observer,
+	)
+	if err != nil {
+		t.Fatalf("NewVideoWorker() error = %v", err)
+	}
+	worker.wait = func(context.Context, time.Duration) error { return nil }
+	return worker
+}
+
+func TestNewVideoWorkerValidatesDependencies(t *testing.T) {
+	factory := &scriptedVideoFactory{}
+	sink := &fakeVideoSink{}
+	retry := testRetryPolicy(3)
+	decide := func(error) bool { return true }
+
+	tests := []struct {
+		name        string
+		factory     VideoReaderFactory
+		sink        VideoSink
+		retry       RetryPolicy
+		shouldRetry func(error) bool
+		wantErr     error
+	}{
+		{
+			name:        "missing factory",
+			sink:        sink,
+			retry:       retry,
+			shouldRetry: decide,
+			wantErr:     ErrVideoFactoryRequired,
+		},
+		{
+			name:        "missing sink",
+			factory:     factory,
+			retry:       retry,
+			shouldRetry: decide,
+			wantErr:     ErrVideoSinkRequired,
+		},
+		{
+			name:    "missing retry decider",
+			factory: factory,
+			sink:    sink,
+			retry:   retry,
+			wantErr: ErrVideoRetryDeciderRequired,
+		},
+		{
+			name:        "invalid retry policy",
+			factory:     factory,
+			sink:        sink,
+			retry:       RetryPolicy{},
+			shouldRetry: decide,
+			wantErr:     ErrInvalidRetryDelay,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			worker, err := NewVideoWorker(
+				tt.factory,
+				tt.sink,
+				tt.retry,
+				tt.shouldRetry,
+				nil,
+			)
+			if worker != nil {
+				t.Fatal("NewVideoWorker() worker is not nil")
+			}
+			if !errors.Is(err, tt.wantErr) {
+				t.Fatalf("NewVideoWorker() error = %v, want %v", err, tt.wantErr)
+			}
+		})
+	}
+}
+
+func TestVideoWorkerRejectsInactiveFeed(t *testing.T) {
+	factory := &scriptedVideoFactory{}
+	var statuses []Status
+	worker := newTestVideoWorker(
+		t,
+		factory,
+		&fakeVideoSink{},
+		3,
+		func(error) bool { return true },
+		func(status Status) { statuses = append(statuses, status) },
+	)
+	config := activeVideoConfig()
+	config.Active = false
+
+	err := worker.Run(context.Background(), config)
+	if !errors.Is(err, ErrVideoFeedInactive) {
+		t.Fatalf("Run() error = %v, want %v", err, ErrVideoFeedInactive)
+	}
+	if factory.calls != 0 {
+		t.Errorf("factory calls = %d, want 0", factory.calls)
+	}
+	if len(statuses) != 0 {
+		t.Errorf("status count = %d, want 0", len(statuses))
+	}
+}
+
+func TestVideoWorkerExhaustsOpenRetries(t *testing.T) {
+	openErr := errors.New("producer unavailable")
+	factory := &scriptedVideoFactory{
+		results: []videoOpenResult{
+			{err: openErr},
+			{err: openErr},
+		},
+	}
+	var statuses []Status
+	worker := newTestVideoWorker(
+		t,
+		factory,
+		&fakeVideoSink{},
+		2,
+		func(error) bool { return true },
+		func(status Status) { statuses = append(statuses, status) },
+	)
+
+	err := worker.Run(context.Background(), activeVideoConfig())
+	if !errors.Is(err, openErr) {
+		t.Fatalf("Run() error = %v, want %v", err, openErr)
+	}
+	if factory.calls != 2 {
+		t.Errorf("factory calls = %d, want 2", factory.calls)
+	}
+
+	wantStates := []State{
+		StateConnecting,
+		StateReconnecting,
+		StateReconnecting,
+		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].State != want {
+			t.Errorf("status %d state = %v, want %v", i, statuses[i].State, want)
+		}
+		if statuses[i].Unit != UnitVideo {
+			t.Errorf("status %d unit = %v, want %v", i, statuses[i].Unit, UnitVideo)
+		}
+	}
+	final := statuses[len(statuses)-1]
+	if final.Attempt != 2 || final.FailedAttempts != 2 {
+		t.Errorf("final status = %+v, want attempt=2 failedAttempts=2", final)
+	}
+	if !errors.Is(final.Err, openErr) {
+		t.Errorf("final error = %v, want %v", final.Err, openErr)
+	}
+}
+
+func TestVideoWorkerStablePlaybackResetsRetryCounter(t *testing.T) {
+	readErr := errors.New("video disconnected")
+	ctx, cancel := context.WithCancel(context.Background())
+	first := &fakeVideoReader{
+		frames:  []VideoFrame{{Index: 1, Payload: []byte{1}}},
+		readErr: readErr,
+	}
+	second := &fakeVideoReader{
+		frames:  []VideoFrame{{Index: 2, Payload: []byte{2}}},
+		readErr: readErr,
+	}
+	third := &fakeVideoReader{
+		read: func(ctx context.Context) (VideoFrame, error) {
+			cancel()
+			return VideoFrame{}, ctx.Err()
+		},
+	}
+	factory := &scriptedVideoFactory{
+		results: []videoOpenResult{
+			{reader: first},
+			{reader: second},
+			{reader: third},
+		},
+	}
+	var statuses []Status
+	worker := newTestVideoWorker(
+		t,
+		factory,
+		&fakeVideoSink{},
+		2,
+		func(error) bool { return true },
+		func(status Status) { statuses = append(statuses, status) },
+	)
+
+	err := worker.Run(ctx, activeVideoConfig())
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("Run() error = %v, want context.Canceled", err)
+	}
+	if factory.calls != 3 {
+		t.Fatalf("factory calls = %d, want 3", factory.calls)
+	}
+
+	var playingAttempts []int
+	var retryFailures []int
+	for _, status := range statuses {
+		switch status.State {
+		case StatePlaying:
+			playingAttempts = append(playingAttempts, status.Attempt)
+		case StateReconnecting:
+			if status.RetryIn > 0 {
+				retryFailures = append(retryFailures, status.FailedAttempts)
+			}
+		}
+	}
+	if len(playingAttempts) != 2 || playingAttempts[0] != 1 || playingAttempts[1] != 2 {
+		t.Errorf("playing attempts = %v, want [1 2]", playingAttempts)
+	}
+	if len(retryFailures) != 2 || retryFailures[0] != 1 || retryFailures[1] != 1 {
+		t.Errorf("retry failure counts = %v, want [1 1]", retryFailures)
+	}
+	wantEnding := []State{StateStopping, StateIdle}
+	if len(statuses) < 2 {
+		t.Fatalf("status count = %d, want at least 2", len(statuses))
+	}
+	ending := statuses[len(statuses)-2:]
+	for i, want := range wantEnding {
+		if ending[i].State != want {
+			t.Errorf("ending status %d = %v, want %v", i, ending[i].State, want)
+		}
+	}
+}
+
+func TestVideoWorkerDoesNotRetrySinkFailure(t *testing.T) {
+	sinkErr := errors.New("renderer failed")
+	reader := &fakeVideoReader{
+		frames: []VideoFrame{{Index: 1, Payload: []byte{1}}},
+	}
+	factory := &scriptedVideoFactory{
+		results: []videoOpenResult{{reader: reader}},
+	}
+	deciderCalls := 0
+	var statuses []Status
+	worker := newTestVideoWorker(
+		t,
+		factory,
+		&fakeVideoSink{err: sinkErr},
+		0,
+		func(error) bool {
+			deciderCalls++
+			return true
+		},
+		func(status Status) { statuses = append(statuses, status) },
+	)
+
+	err := worker.Run(context.Background(), activeVideoConfig())
+	if !errors.Is(err, sinkErr) {
+		t.Fatalf("Run() error = %v, want %v", err, sinkErr)
+	}
+	if factory.calls != 1 {
+		t.Errorf("factory calls = %d, want 1", factory.calls)
+	}
+	if deciderCalls != 0 {
+		t.Errorf("source retry decider calls = %d, want 0", deciderCalls)
+	}
+	if !reader.closed {
+		t.Fatal("reader was not closed")
+	}
+	if len(statuses) != 2 {
+		t.Fatalf("status count = %d, want 2: %+v", len(statuses), statuses)
+	}
+	if statuses[0].State != StateConnecting || statuses[1].State != StateFailed {
+		t.Errorf("status states = [%v %v], want [Connecting Failed]", statuses[0].State, statuses[1].State)
+	}
+}

internal/renderer/renderer.go

@@ -3,17 +3,20 @@ package renderer
 import (
 	"errors"
 	"fmt"
-	"mxl-player/internal/sdl"
 	"unsafe"
 
+	"mxl-player/internal/sdl"
+
 	"github.com/christerso/vulkan-go/vk"
 )
 
-// Raises by swapchain creation, whem windos is minimized
+var (
-var ErrMinimized = errors.New("window minimized")
+	// Raises by swapchain creation, whem windos is minimized
-
+	ErrMinimized = errors.New("window minimized")
-// ErrOutOfDate is returned by DrawFrame when the swapchain needs recreation.
+	// ErrOutOfDate is returned by DrawFrame when the swapchain needs recreation.
-var ErrOutOfDate = errors.New("swapchain out of date")
+	ErrOutOfDate         = errors.New("swapchain out of date")
+	ErrInvalidVideoFrame = errors.New("invalid video frame")
+)
 
 // shader push-constants block
 type PushConstants struct {
@@ -392,6 +395,64 @@ func (r *Renderer) RecreateBuffers(newSize vk.DeviceSize) error {
 	return nil
 }
 
+func (r *Renderer) StageFrame(
+	payload []byte,
+	width uint32,
+	height uint32,
+	stride uint32,
+) error {
+	frameSize, err := validateFramePayload(len(payload), width, height, stride)
+	if err != nil {
+		return err
+	}
+
+	if frameSize != r.FrameSize() {
+		// RecreateBuffers waits for the device to become idle.
+		if err := r.RecreateBuffers(frameSize); err != nil {
+			return fmt.Errorf("resize video buffers: %w", err)
+		}
+	} else {
+		// The previous submitted frame may still read the mapped staging buffer.
+		if err := r.dev.WaitFence(r.inFlight, ^uint64(0)); err != nil {
+			return fmt.Errorf("wait before staging video: %w", err)
+		}
+	}
+
+	vk.CopyToMapped(
+		r.StagingMapped(),
+		payload[:int(frameSize)],
+	)
+	return nil
+}
+
+func validateFramePayload(
+	payloadLen int,
+	width uint32,
+	height uint32,
+	stride uint32,
+) (vk.DeviceSize, error) {
+	if width == 0 || height == 0 || stride == 0 {
+		return 0, fmt.Errorf(
+			"%w: dimensions=%dx%d stride=%d",
+			ErrInvalidVideoFrame,
+			width,
+			height,
+			stride,
+		)
+	}
+
+	requiredSize := uint64(stride) * uint64(height)
+	if payloadLen < 0 || requiredSize > uint64(payloadLen) {
+		return 0, fmt.Errorf(
+			"%w: payload=%d required=%d",
+			ErrInvalidVideoFrame,
+			payloadLen,
+			requiredSize,
+		)
+	}
+	return vk.DeviceSize(requiredSize), nil
+}
+
 // DrawFrame acquires an image, records commands, submits, and presents.
 // Returns ErrOutOfDate if the swapchain needs recreation
 func (r *Renderer) DrawFrame(videoW, videoH, stride uint32) error {

internal/renderer/renderer_test.go

@@ -0,0 +1,122 @@
+package renderer
+
+import (
+	"errors"
+	"strconv"
+	"testing"
+
+	"github.com/christerso/vulkan-go/vk"
+)
+
+func TestValidateFramePayload(t *testing.T) {
+	tests := []struct {
+		name       string
+		payloadLen int
+		width      uint32
+		height     uint32
+		stride     uint32
+		wantSize   vk.DeviceSize
+		wantErr    bool
+	}{
+		{
+			name:       "valid frame",
+			payloadLen: 5120 * 1080,
+			width:      1920,
+			height:     1080,
+			stride:     5120,
+			wantSize:   vk.DeviceSize(5120 * 1080),
+		},
+		{
+			name:       "payload may be larger than frame",
+			payloadLen: 5120*1080 + 128,
+			width:      1920,
+			height:     1080,
+			stride:     5120,
+			wantSize:   vk.DeviceSize(5120 * 1080),
+		},
+		{
+			name:       "zero width",
+			payloadLen: 100,
+			height:     10,
+			stride:     10,
+			wantErr:    true,
+		},
+		{
+			name:       "zero height",
+			payloadLen: 100,
+			width:      10,
+			stride:     10,
+			wantErr:    true,
+		},
+		{
+			name:       "zero stride",
+			payloadLen: 100,
+			width:      10,
+			height:     10,
+			wantErr:    true,
+		},
+		{
+			name:       "payload is too small",
+			payloadLen: 99,
+			width:      10,
+			height:     10,
+			stride:     10,
+			wantErr:    true,
+		},
+		{
+			name:       "negative payload length",
+			payloadLen: -1,
+			width:      10,
+			height:     10,
+			stride:     10,
+			wantErr:    true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := validateFramePayload(
+				tt.payloadLen,
+				tt.width,
+				tt.height,
+				tt.stride,
+			)
+
+			if tt.wantErr {
+				if !errors.Is(err, ErrInvalidVideoFrame) {
+					t.Fatalf("validateFramePayload() error = %v, want %v", err, ErrInvalidVideoFrame)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("validateFramePayload() error = %v, want nil", err)
+			}
+			if got != tt.wantSize {
+				t.Errorf("validateFramePayload() size = %d, want %d", got, tt.wantSize)
+			}
+		})
+	}
+}
+
+func TestValidateFramePayloadUses64BitSize(t *testing.T) {
+	if strconv.IntSize < 64 {
+		t.Skip("test requires a 64-bit int")
+	}
+
+	stride := ^uint32(0)
+	height := uint32(2)
+	required := uint64(stride) * uint64(height)
+
+	got, err := validateFramePayload(
+		int(required),
+		1,
+		height,
+		stride,
+	)
+	if err != nil {
+		t.Fatalf("validateFramePayload() error = %v, want nil", err)
+	}
+	if uint64(got) != required {
+		t.Fatalf("validateFramePayload() size = %d, want %d", got, required)
+	}
+}

internal/source/source.go

@@ -120,15 +120,43 @@ func (s *Source) Close() error {
 
 func (s *Source) NextCtx(ctx context.Context, timeout time.Duration) (Frame, error) {
 	for {
+		frame, err := s.ReadOnceCtx(ctx, timeout)
+		if err == nil {
+			return frame, nil
+		}
+		if ctx.Err() != nil {
+			return Frame{}, ctx.Err()
+		}
+		if KindOf(err) != ErrorKindTemporary {
+			return Frame{}, err
+		}
+
+		if errors.Is(err, mxl.ErrOutOfRangeEarly) {
 			select {
+			case <-time.After(10 * time.Millisecond):
 			case <-ctx.Done():
 				return Frame{}, ctx.Err()
-		default:
 			}
+		}
+	}
+}
+
+func (s *Source) ReadOnceCtx(
+	ctx context.Context,
+	timeout time.Duration,
+) (Frame, error) {
+	if err := ctx.Err(); err != nil {
+		return Frame{}, err
+	}
+
 	g, err := s.reader.GetGrain(s.idx, timeout)
+	if ctx.Err() != nil {
+		return Frame{}, ctx.Err()
+	}
+
 	switch {
 	case err == nil:
-			f := Frame{
+		frame := Frame{
 			Index:   g.Index,
 			Width:   s.width,
 			Height:  s.height,
@@ -138,16 +166,44 @@ func (s *Source) NextCtx(ctx context.Context, timeout time.Duration) (Frame, err
 			Payload: g.Payload,
 		}
 		s.idx++
-			return f, nil
+		return frame, nil
+
 	case errors.Is(err, mxl.ErrTimeout):
 		s.idx = mxl.CurrentIndex(s.rate)
+		return Frame{}, wrapError(
+			"read video",
+			ErrorKindTemporary,
+			err,
+		)
+
 	case errors.Is(err, mxl.ErrOutOfRangeEarly):
-			time.Sleep(10 * time.Millisecond)
+		return Frame{}, wrapError(
+			"read video",
+			ErrorKindTemporary,
+			err,
+		)
+
 	case errors.Is(err, mxl.ErrOutOfRangeLate):
 		s.idx = mxl.CurrentIndex(s.rate)
+		return Frame{}, wrapError(
+			"read video",
+			ErrorKindTemporary,
+			err,
+		)
+
+	case errors.Is(err, mxl.ErrFlowInvalid):
+		return Frame{}, wrapError(
+			"read video",
+			ErrorKindUnavailable,
+			err,
+		)
+
 	default:
-			return Frame{}, fmt.Errorf("GetGrain: %w", err)
+		return Frame{}, wrapError(
-		}
+			"read video",
+			ErrorKindUnavailable,
+			err,
+		)
 	}
 }