add independent video worker

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)
+	}
+}