detect prolonged video unavailability

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