add local MXL video adapter

internal/adapter/mxl/retry.go

@@ -0,0 +1,26 @@
+package mxladapter
+
+import (
+	"context"
+	"errors"
+
+	"mxl-player/internal/source"
+)
+
+// ShouldRetry reports whether an MXL source error should start another attempt.
+func ShouldRetry(err error) bool {
+	if errors.Is(err, context.Canceled) {
+		return false
+	}
+
+	switch source.KindOf(err) {
+	case source.ErrorKindInvalidConfig:
+		return false
+	case source.ErrorKindTemporary,
+		source.ErrorKindUnavailable,
+		source.ErrorKindUnknown:
+		return true
+	default:
+		return true
+	}
+}

internal/adapter/mxl/retry_test.go

@@ -0,0 +1,74 @@
+package mxladapter
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"testing"
+
+	"mxl-player/internal/source"
+)
+
+func TestShouldRetrySourceError(t *testing.T) {
+	baseErr := errors.New("source failed")
+	classified := func(kind source.ErrorKind) error {
+		return &source.SourceError{
+			Op:   "read media",
+			Kind: kind,
+			Err:  baseErr,
+		}
+	}
+
+	tests := []struct {
+		name string
+		err  error
+		want bool
+	}{
+		{
+			name: "temporary source error is retryable",
+			err:  classified(source.ErrorKindTemporary),
+			want: true,
+		},
+		{
+			name: "unavailable source is retryable",
+			err:  classified(source.ErrorKindUnavailable),
+			want: true,
+		},
+		{
+			name: "invalid configuration is not retryable",
+			err:  classified(source.ErrorKindInvalidConfig),
+			want: false,
+		},
+		{
+			name: "ordinary unknown error is retryable",
+			err:  baseErr,
+			want: true,
+		},
+		{
+			name: "wrapped invalid configuration is not retryable",
+			err: fmt.Errorf(
+				"worker failed: %w",
+				classified(source.ErrorKindInvalidConfig),
+			),
+			want: false,
+		},
+		{
+			name: "context cancellation is not retryable",
+			err:  context.Canceled,
+			want: false,
+		},
+		{
+			name: "wrapped context cancellation is not retryable",
+			err:  fmt.Errorf("worker stopped: %w", context.Canceled),
+			want: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := ShouldRetry(tt.err); got != tt.want {
+				t.Errorf("ShouldRetry() = %t, want %t", got, tt.want)
+			}
+		})
+	}
+}

internal/adapter/mxl/video.go

@@ -0,0 +1,91 @@
+package mxladapter
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"mxl-player/internal/playback"
+	"mxl-player/internal/source"
+)
+
+const DefaultVideoReadTimeout = 200 * time.Millisecond
+
+type VideoFactory struct {
+	ReadTimeout time.Duration
+}
+
+type videoReader struct {
+	source  *source.Source
+	timeout time.Duration
+}
+
+var _ playback.VideoReaderFactory = VideoFactory{}
+var _ playback.VideoReader = (*videoReader)(nil)
+
+func (f VideoFactory) OpenVideo(
+	ctx context.Context,
+	config playback.FeedConfig,
+) (playback.VideoReader, error) {
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
+	if err := config.Validate(); err != nil {
+		return nil, &source.SourceError{
+			Op:   "validate video feed",
+			Kind: source.ErrorKindInvalidConfig,
+			Err:  err,
+		}
+	}
+	if !config.IsConfigured() {
+		return nil, &source.SourceError{
+			Op:   "validate video feed",
+			Kind: source.ErrorKindInvalidConfig,
+			Err:  errors.New("video feed is not configured"),
+		}
+	}
+
+	src, err := source.Open(config.Domain, config.UUID)
+	if err != nil {
+		return nil, fmt.Errorf("open local MXL video: %w", err)
+	}
+
+	if err := ctx.Err(); err != nil {
+		_ = src.Close()
+		return nil, err
+	}
+
+	timeout := f.ReadTimeout
+	if timeout <= 0 {
+		timeout = DefaultVideoReadTimeout
+	}
+
+	return &videoReader{
+		source:  src,
+		timeout: timeout,
+	}, nil
+}
+
+func (r *videoReader) ReadVideo(
+	ctx context.Context,
+) (playback.VideoFrame, error) {
+	frame, err := r.source.NextCtx(ctx, r.timeout)
+	if err != nil {
+		return playback.VideoFrame{}, err
+	}
+
+	return playback.VideoFrame{
+		Index:   frame.Index,
+		Width:   frame.Width,
+		Height:  frame.Height,
+		Stride:  frame.Stride,
+		Size:    frame.Size,
+		Invalid: frame.Invalid,
+		Payload: frame.Payload,
+	}, nil
+}
+
+func (r *videoReader) Close() error {
+	return r.source.Close()
+}

internal/adapter/mxl/video_test.go

@@ -0,0 +1,75 @@
+package mxladapter
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"mxl-player/internal/playback"
+	"mxl-player/internal/source"
+)
+
+func TestVideoFactoryOpenVideoCanceled(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	reader, err := (VideoFactory{}).OpenVideo(ctx, playback.FeedConfig{})
+	if reader != nil {
+		t.Fatal("OpenVideo() reader is not nil after cancellation")
+	}
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("OpenVideo() error = %v, want context.Canceled", err)
+	}
+}
+
+func TestVideoFactoryOpenVideoRejectsInvalidConfig(t *testing.T) {
+	tests := []struct {
+		name   string
+		config playback.FeedConfig
+	}{
+		{
+			name:   "feed is not configured",
+			config: playback.FeedConfig{},
+		},
+		{
+			name: "active feed has no UUID",
+			config: playback.FeedConfig{
+				Domain: "/dev/shm/mxl",
+				Active: true,
+			},
+		},
+		{
+			name: "configured feed has no domain",
+			config: playback.FeedConfig{
+				UUID:   "video-uuid",
+				Active: true,
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			reader, err := (VideoFactory{}).OpenVideo(
+				context.Background(),
+				tt.config,
+			)
+
+			if reader != nil {
+				t.Fatal("OpenVideo() reader is not nil for invalid config")
+			}
+			if err == nil {
+				t.Fatal("OpenVideo() error is nil for invalid config")
+			}
+			if got := source.KindOf(err); got != source.ErrorKindInvalidConfig {
+				t.Fatalf(
+					"source.KindOf(OpenVideo()) = %v, want %v",
+					got,
+					source.ErrorKindInvalidConfig,
+				)
+			}
+			if ShouldRetry(err) {
+				t.Fatal("ShouldRetry(OpenVideo()) = true for invalid config")
+			}
+		})
+	}
+}

internal/playback/supervisor.go

@@ -0,0 +1,82 @@
+package playback
+
+import (
+	"context"
+	"time"
+)
+
+type attemptFunc func(context.Context) error
+type retryDecider func(error) bool
+type waitFunc func(context.Context, time.Duration) error
+
+func waitForRetry(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()
+	}
+}
+
+func runWithRetry(
+	ctx context.Context,
+	policy RetryPolicy,
+	attempt attemptFunc,
+	shouldRetry retryDecider,
+	wait waitFunc,
+	observer retryObserver,
+) error {
+	failedAttempts := 0
+
+	for {
+		err := attempt(ctx)
+		if err == nil {
+			return nil
+		}
+		if ctx.Err() != nil {
+			return ctx.Err()
+		}
+
+		failedAttempts++
+		willRetry := shouldRetry(err) && policy.canRetry(failedAttempts)
+		if !willRetry {
+			if observer != nil {
+				observer(retryEvent{
+					FailedAttempts: failedAttempts,
+					Err:            err,
+					WillRetry:      false,
+				})
+			}
+			return err
+		}
+
+		delay := policy.retryDelay(failedAttempts)
+		if observer != nil {
+			observer(retryEvent{
+				FailedAttempts: failedAttempts,
+				Err:            err,
+				RetryIn:        delay,
+				WillRetry:      true,
+			})
+		}
+
+		if err := wait(ctx, delay); err != nil {
+			if ctx.Err() != nil {
+				return ctx.Err()
+			}
+			return err
+		}
+	}
+}
+
+type retryEvent struct {
+	FailedAttempts int
+	Err            error
+	RetryIn        time.Duration
+	WillRetry      bool
+}
+
+type retryObserver func(retryEvent)

internal/playback/supervisor_test.go

@@ -0,0 +1,367 @@
+package playback
+
+import (
+	"context"
+	"errors"
+	"reflect"
+	"testing"
+	"time"
+)
+
+func testRetryPolicy(maxAttempts int) RetryPolicy {
+	return RetryPolicy{
+		MaxAttempts:  maxAttempts,
+		InitialDelay: 500 * time.Millisecond,
+		MaxDelay:     10 * time.Second,
+	}
+}
+
+func TestRunWithRetryFirstAttemptSucceeds(t *testing.T) {
+	attempts := 0
+	err := runWithRetry(
+		context.Background(),
+		testRetryPolicy(3),
+		func(context.Context) error {
+			attempts++
+			return nil
+		},
+		func(error) bool {
+			t.Fatal("shouldRetry called after successful attempt")
+			return false
+		},
+		func(context.Context, time.Duration) error {
+			t.Fatal("wait called after successful attempt")
+			return nil
+		},
+		nil,
+	)
+
+	if err != nil {
+		t.Fatalf("runWithRetry() error = %v, want nil", err)
+	}
+	if attempts != 1 {
+		t.Fatalf("attempt count = %d, want 1", attempts)
+	}
+}
+
+func TestRunWithRetryFailuresThenSuccess(t *testing.T) {
+	attemptErr := errors.New("attempt failed")
+	attempts := 0
+	var delays []time.Duration
+
+	err := runWithRetry(
+		context.Background(),
+		testRetryPolicy(3),
+		func(context.Context) error {
+			attempts++
+			if attempts < 3 {
+				return attemptErr
+			}
+			return nil
+		},
+		func(error) bool { return true },
+		func(_ context.Context, delay time.Duration) error {
+			delays = append(delays, delay)
+			return nil
+		},
+		nil,
+	)
+
+	if err != nil {
+		t.Fatalf("runWithRetry() error = %v, want nil", err)
+	}
+	if attempts != 3 {
+		t.Errorf("attempt count = %d, want 3", attempts)
+	}
+	wantDelays := []time.Duration{500 * time.Millisecond, time.Second}
+	if !reflect.DeepEqual(delays, wantDelays) {
+		t.Errorf("retry delays = %v, want %v", delays, wantDelays)
+	}
+}
+
+func TestRunWithRetryFiniteAttemptsExhausted(t *testing.T) {
+	attemptErr := errors.New("attempt failed")
+	attempts := 0
+	waits := 0
+
+	err := runWithRetry(
+		context.Background(),
+		testRetryPolicy(3),
+		func(context.Context) error {
+			attempts++
+			return attemptErr
+		},
+		func(error) bool { return true },
+		func(context.Context, time.Duration) error {
+			waits++
+			return nil
+		},
+		nil,
+	)
+
+	if !errors.Is(err, attemptErr) {
+		t.Fatalf("runWithRetry() error = %v, want %v", err, attemptErr)
+	}
+	if attempts != 3 {
+		t.Errorf("attempt count = %d, want 3", attempts)
+	}
+	if waits != 2 {
+		t.Errorf("wait count = %d, want 2", waits)
+	}
+}
+
+func TestRunWithRetryUnlimitedEventuallySucceeds(t *testing.T) {
+	attemptErr := errors.New("attempt failed")
+	attempts := 0
+
+	err := runWithRetry(
+		context.Background(),
+		testRetryPolicy(0),
+		func(context.Context) error {
+			attempts++
+			if attempts < 20 {
+				return attemptErr
+			}
+			return nil
+		},
+		func(error) bool { return true },
+		func(context.Context, time.Duration) error { return nil },
+		nil,
+	)
+
+	if err != nil {
+		t.Fatalf("runWithRetry() error = %v, want nil", err)
+	}
+	if attempts != 20 {
+		t.Fatalf("attempt count = %d, want 20", attempts)
+	}
+}
+
+func TestRunWithRetryStopsWhenErrorIsNotRetryable(t *testing.T) {
+	attemptErr := errors.New("invalid configuration")
+	attempts := 0
+
+	err := runWithRetry(
+		context.Background(),
+		testRetryPolicy(0),
+		func(context.Context) error {
+			attempts++
+			return attemptErr
+		},
+		func(error) bool { return false },
+		func(context.Context, time.Duration) error {
+			t.Fatal("wait called for non-retryable error")
+			return nil
+		},
+		nil,
+	)
+
+	if !errors.Is(err, attemptErr) {
+		t.Fatalf("runWithRetry() error = %v, want %v", err, attemptErr)
+	}
+	if attempts != 1 {
+		t.Fatalf("attempt count = %d, want 1", attempts)
+	}
+}
+
+func TestRunWithRetryReturnsCancellationFromAttempt(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	attemptErr := errors.New("attempt failed")
+
+	err := runWithRetry(
+		ctx,
+		testRetryPolicy(0),
+		func(context.Context) error {
+			cancel()
+			return attemptErr
+		},
+		func(error) bool { return true },
+		func(context.Context, time.Duration) error {
+			t.Fatal("wait called after cancellation")
+			return nil
+		},
+		nil,
+	)
+
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("runWithRetry() error = %v, want context.Canceled", err)
+	}
+}
+
+func TestRunWithRetryReturnsCancellationDuringBackoff(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	attemptErr := errors.New("attempt failed")
+
+	err := runWithRetry(
+		ctx,
+		testRetryPolicy(0),
+		func(context.Context) error { return attemptErr },
+		func(error) bool { return true },
+		func(ctx context.Context, _ time.Duration) error {
+			cancel()
+			return ctx.Err()
+		},
+		nil,
+	)
+
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("runWithRetry() error = %v, want context.Canceled", err)
+	}
+}
+
+func TestRunWithRetryReturnsWaitError(t *testing.T) {
+	attemptErr := errors.New("attempt failed")
+	waitErr := errors.New("wait failed")
+
+	err := runWithRetry(
+		context.Background(),
+		testRetryPolicy(0),
+		func(context.Context) error { return attemptErr },
+		func(error) bool { return true },
+		func(context.Context, time.Duration) error { return waitErr },
+		nil,
+	)
+
+	if !errors.Is(err, waitErr) {
+		t.Fatalf("runWithRetry() error = %v, want %v", err, waitErr)
+	}
+}
+
+func TestWaitForRetryReturnsCancellation(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	err := waitForRetry(ctx, time.Hour)
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("waitForRetry() error = %v, want context.Canceled", err)
+	}
+}
+
+func TestRetryObserverReportsFailuresBeforeSuccess(t *testing.T) {
+	attemptErr := errors.New("attempt failed")
+	attempts := 0
+	var events []retryEvent
+
+	err := runWithRetry(
+		context.Background(),
+		testRetryPolicy(3),
+		func(context.Context) error {
+			attempts++
+			if attempts < 3 {
+				return attemptErr
+			}
+			return nil
+		},
+		func(error) bool { return true },
+		func(context.Context, time.Duration) error { return nil },
+		func(event retryEvent) {
+			events = append(events, event)
+		},
+	)
+
+	if err != nil {
+		t.Fatalf("runWithRetry() error = %v, want nil", err)
+	}
+	if len(events) != 2 {
+		t.Fatalf("event count = %d, want 2", len(events))
+	}
+
+	wantDelays := []time.Duration{500 * time.Millisecond, time.Second}
+	for i, event := range events {
+		wantAttempts := i + 1
+		if event.FailedAttempts != wantAttempts {
+			t.Errorf("event %d failed attempts = %d, want %d", i, event.FailedAttempts, wantAttempts)
+		}
+		if !errors.Is(event.Err, attemptErr) {
+			t.Errorf("event %d error = %v, want %v", i, event.Err, attemptErr)
+		}
+		if event.RetryIn != wantDelays[i] {
+			t.Errorf("event %d retry delay = %s, want %s", i, event.RetryIn, wantDelays[i])
+		}
+		if !event.WillRetry {
+			t.Errorf("event %d WillRetry = false, want true", i)
+		}
+	}
+}
+
+func TestRetryObserverReportsExhaustion(t *testing.T) {
+	attemptErr := errors.New("attempt failed")
+	var events []retryEvent
+
+	err := runWithRetry(
+		context.Background(),
+		testRetryPolicy(2),
+		func(context.Context) error { return attemptErr },
+		func(error) bool { return true },
+		func(context.Context, time.Duration) error { return nil },
+		func(event retryEvent) {
+			events = append(events, event)
+		},
+	)
+
+	if !errors.Is(err, attemptErr) {
+		t.Fatalf("runWithRetry() error = %v, want %v", err, attemptErr)
+	}
+	if len(events) != 2 {
+		t.Fatalf("event count = %d, want 2", len(events))
+	}
+	if !events[0].WillRetry || events[0].RetryIn != 500*time.Millisecond {
+		t.Errorf("first event = %+v, want retry after 500ms", events[0])
+	}
+	final := events[1]
+	if final.FailedAttempts != 2 {
+		t.Errorf("final failed attempts = %d, want 2", final.FailedAttempts)
+	}
+	if final.WillRetry {
+		t.Error("final WillRetry = true, want false")
+	}
+	if final.RetryIn != 0 {
+		t.Errorf("final retry delay = %s, want 0", final.RetryIn)
+	}
+	if !errors.Is(final.Err, attemptErr) {
+		t.Errorf("final error = %v, want %v", final.Err, attemptErr)
+	}
+}
+
+func TestRetryObserverNotCalledOnImmediateSuccess(t *testing.T) {
+	observerCalls := 0
+	err := runWithRetry(
+		context.Background(),
+		testRetryPolicy(3),
+		func(context.Context) error { return nil },
+		func(error) bool { return true },
+		func(context.Context, time.Duration) error { return nil },
+		func(retryEvent) { observerCalls++ },
+	)
+
+	if err != nil {
+		t.Fatalf("runWithRetry() error = %v, want nil", err)
+	}
+	if observerCalls != 0 {
+		t.Fatalf("observer call count = %d, want 0", observerCalls)
+	}
+}
+
+func TestRetryObserverNotCalledWhenAttemptCancelsContext(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	observerCalls := 0
+
+	err := runWithRetry(
+		ctx,
+		testRetryPolicy(0),
+		func(context.Context) error {
+			cancel()
+			return errors.New("attempt interrupted")
+		},
+		func(error) bool { return true },
+		func(context.Context, time.Duration) error { return nil },
+		func(retryEvent) { observerCalls++ },
+	)
+
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("runWithRetry() error = %v, want context.Canceled", err)
+	}
+	if observerCalls != 0 {
+		t.Fatalf("observer call count = %d, want 0", observerCalls)
+	}
+}

internal/playback/video.go

@@ -0,0 +1,40 @@
+package playback
+
+import "context"
+
+// VideoFrame contains metadata and borrowed source payload.
+//
+// Payload is valid only until the next VideoReader.ReadVideo call or until the
+// reader is closed. Consumers must finish reading Payload before returning
+// control to the worker
+type VideoFrame struct {
+	Index   uint64
+	Width   uint32
+	Height  uint32
+	Stride  uint32
+	Size    uint32
+	Invalid bool
+	Payload []byte
+}
+
+// VideoReader reads frames from a video source.
+//
+// ReadVideo must not be called again until the previous frame's payload has
+// been consumed.
+type VideoReader interface {
+	ReadVideo(context.Context) (VideoFrame, error)
+	Close() error
+}
+
+// VideoReaderFactory opens a reader for the configured video feed.
+type VideoReaderFactory interface {
+	OpenVideo(context.Context, FeedConfig) (VideoReader, error)
+}
+
+// VideoSink consumes a borrowed video frame.
+//
+// ConsumeVideo must finish using frame.Payload before returning and must never
+// retain it for asynchronous use.
+type VideoSink interface {
+	ConsumeVideo(context.Context, VideoFrame) error
+}