From 862f76c1b8796466bda56a9f94d6ffcbe2cca51e Mon Sep 17 00:00:00 2001 From: Dmitry Sergeev Date: Thu, 27 Aug 2026 10:23:57 +0300 Subject: [PATCH] add independent video worker --- internal/playback/video_worker.go | 184 ++++++++++++++ internal/playback/video_worker_test.go | 319 +++++++++++++++++++++++++ 2 files changed, 503 insertions(+) create mode 100644 internal/playback/video_worker.go create mode 100644 internal/playback/video_worker_test.go diff --git a/internal/playback/video_worker.go b/internal/playback/video_worker.go new file mode 100644 index 0000000..bc929a6 --- /dev/null +++ b/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 +} diff --git a/internal/playback/video_worker_test.go b/internal/playback/video_worker_test.go new file mode 100644 index 0000000..30a96b9 --- /dev/null +++ b/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) + } +}