reset retries after stable playback

internal/playback/state.go

@@ -1,5 +1,15 @@
 package playback
 
+import "time"
+
+type Unit uint8
+
+const (
+	UnitVideo Unit = iota
+	UnitAudio
+	UnitSync
+)
+
 type State uint8
 
 const (
@@ -10,3 +20,14 @@ const (
 	StateFailed
 	StateStopping
 )
+
+type Status struct {
+	Unit           Unit
+	State          State
+	Attempt        int
+	FailedAttempts int
+	RetryIn        time.Duration
+	Err            error
+}
+
+type StatusObserver func(Status)

internal/playback/state_test.go

@@ -0,0 +1,40 @@
+package playback
+
+import (
+	"errors"
+	"testing"
+	"time"
+)
+
+func TestStatusPreservesValues(t *testing.T) {
+	wantErr := errors.New("producer missing")
+	status := Status{
+		Unit:           UnitVideo,
+		State:          StateReconnecting,
+		Attempt:        2,
+		FailedAttempts: 1,
+		RetryIn:        time.Second,
+		Err:            wantErr,
+	}
+
+	if status.Unit != UnitVideo {
+		t.Errorf("Unit = %v, want %v", status.Unit, UnitVideo)
+	}
+
+	if !errors.Is(status.Err, wantErr) {
+		t.Errorf("Err = %v, want %v", status.Err, wantErr)
+	}
+
+	if status.State != StateReconnecting {
+		t.Errorf("State = %v, want %v", status.State, StateReconnecting)
+	}
+	if status.Attempt != 2 {
+		t.Errorf("Attempt = %d, want 2", status.Attempt)
+	}
+	if status.FailedAttempts != 1 {
+		t.Errorf("FailedAttempts = %d, want 1", status.FailedAttempts)
+	}
+	if status.RetryIn != time.Second {
+		t.Errorf("RetryIn = %s, want %s", status.RetryIn, time.Second)
+	}
+}

internal/playback/supervisor.go

@@ -5,7 +5,8 @@ import (
 	"time"
 )
 
-type attemptFunc func(context.Context) error
+// attemptFunc returns whether useful media was received before the attempt ended.
+type attemptFunc func(context.Context) (becameStable bool, err error)
 type retryDecider func(error) bool
 type waitFunc func(context.Context, time.Duration) error
 
@@ -32,7 +33,7 @@ func runWithRetry(
 	failedAttempts := 0
 
 	for {
-		err := attempt(ctx)
+		becameStable, err := attempt(ctx)
 		if err == nil {
 			return nil
 		}
@@ -40,6 +41,10 @@ func runWithRetry(
 			return ctx.Err()
 		}
 
+		if becameStable {
+			failedAttempts = 0
+		}
+
 		failedAttempts++
 		willRetry := shouldRetry(err) && policy.canRetry(failedAttempts)
 		if !willRetry {

internal/playback/supervisor_test.go

@@ -21,9 +21,9 @@ func TestRunWithRetryFirstAttemptSucceeds(t *testing.T) {
 	err := runWithRetry(
 		context.Background(),
 		testRetryPolicy(3),
-		func(context.Context) error {
+		func(context.Context) (bool, error) {
 			attempts++
-			return nil
+			return false, nil
 		},
 		func(error) bool {
 			t.Fatal("shouldRetry called after successful attempt")
@@ -52,12 +52,12 @@ func TestRunWithRetryFailuresThenSuccess(t *testing.T) {
 	err := runWithRetry(
 		context.Background(),
 		testRetryPolicy(3),
-		func(context.Context) error {
+		func(context.Context) (bool, error) {
 			attempts++
 			if attempts < 3 {
-				return attemptErr
+				return false, attemptErr
 			}
-			return nil
+			return false, nil
 		},
 		func(error) bool { return true },
 		func(_ context.Context, delay time.Duration) error {
@@ -87,9 +87,9 @@ func TestRunWithRetryFiniteAttemptsExhausted(t *testing.T) {
 	err := runWithRetry(
 		context.Background(),
 		testRetryPolicy(3),
-		func(context.Context) error {
+		func(context.Context) (bool, error) {
 			attempts++
-			return attemptErr
+			return false, attemptErr
 		},
 		func(error) bool { return true },
 		func(context.Context, time.Duration) error {
@@ -117,12 +117,12 @@ func TestRunWithRetryUnlimitedEventuallySucceeds(t *testing.T) {
 	err := runWithRetry(
 		context.Background(),
 		testRetryPolicy(0),
-		func(context.Context) error {
+		func(context.Context) (bool, error) {
 			attempts++
 			if attempts < 20 {
-				return attemptErr
+				return false, attemptErr
 			}
-			return nil
+			return false, nil
 		},
 		func(error) bool { return true },
 		func(context.Context, time.Duration) error { return nil },
@@ -144,9 +144,9 @@ func TestRunWithRetryStopsWhenErrorIsNotRetryable(t *testing.T) {
 	err := runWithRetry(
 		context.Background(),
 		testRetryPolicy(0),
-		func(context.Context) error {
+		func(context.Context) (bool, error) {
 			attempts++
-			return attemptErr
+			return false, attemptErr
 		},
 		func(error) bool { return false },
 		func(context.Context, time.Duration) error {
@@ -171,9 +171,9 @@ func TestRunWithRetryReturnsCancellationFromAttempt(t *testing.T) {
 	err := runWithRetry(
 		ctx,
 		testRetryPolicy(0),
-		func(context.Context) error {
+		func(context.Context) (bool, error) {
 			cancel()
-			return attemptErr
+			return false, attemptErr
 		},
 		func(error) bool { return true },
 		func(context.Context, time.Duration) error {
@@ -195,7 +195,7 @@ func TestRunWithRetryReturnsCancellationDuringBackoff(t *testing.T) {
 	err := runWithRetry(
 		ctx,
 		testRetryPolicy(0),
-		func(context.Context) error { return attemptErr },
+		func(context.Context) (bool, error) { return false, attemptErr },
 		func(error) bool { return true },
 		func(ctx context.Context, _ time.Duration) error {
 			cancel()
@@ -216,7 +216,7 @@ func TestRunWithRetryReturnsWaitError(t *testing.T) {
 	err := runWithRetry(
 		context.Background(),
 		testRetryPolicy(0),
-		func(context.Context) error { return attemptErr },
+		func(context.Context) (bool, error) { return false, attemptErr },
 		func(error) bool { return true },
 		func(context.Context, time.Duration) error { return waitErr },
 		nil,
@@ -245,12 +245,12 @@ func TestRetryObserverReportsFailuresBeforeSuccess(t *testing.T) {
 	err := runWithRetry(
 		context.Background(),
 		testRetryPolicy(3),
-		func(context.Context) error {
+		func(context.Context) (bool, error) {
 			attempts++
 			if attempts < 3 {
-				return attemptErr
+				return false, attemptErr
 			}
-			return nil
+			return false, nil
 		},
 		func(error) bool { return true },
 		func(context.Context, time.Duration) error { return nil },
@@ -291,7 +291,7 @@ func TestRetryObserverReportsExhaustion(t *testing.T) {
 	err := runWithRetry(
 		context.Background(),
 		testRetryPolicy(2),
-		func(context.Context) error { return attemptErr },
+		func(context.Context) (bool, error) { return false, attemptErr },
 		func(error) bool { return true },
 		func(context.Context, time.Duration) error { return nil },
 		func(event retryEvent) {
@@ -328,7 +328,7 @@ func TestRetryObserverNotCalledOnImmediateSuccess(t *testing.T) {
 	err := runWithRetry(
 		context.Background(),
 		testRetryPolicy(3),
-		func(context.Context) error { return nil },
+		func(context.Context) (bool, error) { return false, nil },
 		func(error) bool { return true },
 		func(context.Context, time.Duration) error { return nil },
 		func(retryEvent) { observerCalls++ },
@@ -349,9 +349,9 @@ func TestRetryObserverNotCalledWhenAttemptCancelsContext(t *testing.T) {
 	err := runWithRetry(
 		ctx,
 		testRetryPolicy(0),
-		func(context.Context) error {
+		func(context.Context) (bool, error) {
 			cancel()
-			return errors.New("attempt interrupted")
+			return false, errors.New("attempt interrupted")
 		},
 		func(error) bool { return true },
 		func(context.Context, time.Duration) error { return nil },
@@ -365,3 +365,61 @@ func TestRetryObserverNotCalledWhenAttemptCancelsContext(t *testing.T) {
 		t.Fatalf("observer call count = %d, want 0", observerCalls)
 	}
 }
+
+func TestRunWithRetryResetsFailuresAfterStableAttempt(t *testing.T) {
+	attemptErr := errors.New("attempt failed")
+	attempts := 0
+	var events []retryEvent
+
+	err := runWithRetry(
+		context.Background(),
+		testRetryPolicy(2),
+		func(context.Context) (bool, error) {
+			attempts++
+			switch attempts {
+			case 1:
+				return false, attemptErr
+			case 2:
+				return true, attemptErr
+			default:
+				return false, 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 attempts != 3 {
+		t.Fatalf("attempt count = %d, want 3", attempts)
+	}
+	if len(events) != 3 {
+		t.Fatalf("event count = %d, want 3", len(events))
+	}
+
+	wantFailedAttempts := []int{1, 1, 2}
+	wantWillRetry := []bool{true, true, false}
+	for i, event := range events {
+		if event.FailedAttempts != wantFailedAttempts[i] {
+			t.Errorf(
+				"event %d failed attempts = %d, want %d",
+				i,
+				event.FailedAttempts,
+				wantFailedAttempts[i],
+			)
+		}
+		if event.WillRetry != wantWillRetry[i] {
+			t.Errorf(
+				"event %d WillRetry = %t, want %t",
+				i,
+				event.WillRetry,
+				wantWillRetry[i],
+			)
+		}
+	}
+}

internal/playback/video_attempt.go

@@ -0,0 +1,55 @@
+package playback
+
+import (
+	"context"
+	"errors"
+	"fmt"
+)
+
+type videoSinkError struct {
+	err error
+}
+
+func (e *videoSinkError) Error() string {
+	return fmt.Sprintf("consume video: %v", e.err)
+}
+
+func (e *videoSinkError) Unwrap() error {
+	return e.err
+}
+
+func runVideoAttempt(
+	ctx context.Context,
+	factory VideoReaderFactory,
+	sink VideoSink,
+	config FeedConfig,
+) (resultErr error) {
+	reader, err := factory.OpenVideo(ctx, config)
+	if err != nil {
+		return fmt.Errorf("open video: %w", err)
+	}
+
+	defer func() {
+		if closeErr := reader.Close(); closeErr != nil {
+			closeErr = fmt.Errorf("close video: %w", closeErr)
+			resultErr = errors.Join(resultErr, closeErr)
+		}
+	}()
+
+	for {
+		frame, err := reader.ReadVideo(ctx)
+		if err != nil {
+			if ctx.Err() != nil {
+				return ctx.Err()
+			}
+			return fmt.Errorf("read video: %w", err)
+		}
+
+		if err := sink.ConsumeVideo(ctx, frame); err != nil {
+			if ctx.Err() != nil {
+				return ctx.Err()
+			}
+			return &videoSinkError{err: err}
+		}
+	}
+}

internal/playback/video_attempt_test.go

@@ -0,0 +1,213 @@
+package playback
+
+import (
+	"context"
+	"errors"
+	"testing"
+)
+
+type fakeVideoFactory struct {
+	reader VideoReader
+	err    error
+	calls  int
+}
+
+func (f *fakeVideoFactory) OpenVideo(
+	context.Context,
+	FeedConfig,
+) (VideoReader, error) {
+	f.calls++
+	return f.reader, f.err
+}
+
+type fakeVideoReader struct {
+	frames    []VideoFrame
+	readErr   error
+	closeErr  error
+	readCalls int
+	closed    bool
+	read      func(context.Context) (VideoFrame, error)
+}
+
+func (r *fakeVideoReader) ReadVideo(ctx context.Context) (VideoFrame, error) {
+	r.readCalls++
+	if r.read != nil {
+		return r.read(ctx)
+	}
+	if len(r.frames) == 0 {
+		return VideoFrame{}, r.readErr
+	}
+	frame := r.frames[0]
+	r.frames = r.frames[1:]
+	return frame, nil
+}
+
+func (r *fakeVideoReader) Close() error {
+	r.closed = true
+	return r.closeErr
+}
+
+type fakeVideoSink struct {
+	frames []VideoFrame
+	err    error
+}
+
+func (s *fakeVideoSink) ConsumeVideo(_ context.Context, frame VideoFrame) error {
+	s.frames = append(s.frames, frame)
+	return s.err
+}
+
+func TestRunVideoAttemptOpenFailure(t *testing.T) {
+	openErr := errors.New("open failed")
+	factory := &fakeVideoFactory{err: openErr}
+	sink := &fakeVideoSink{}
+
+	err := runVideoAttempt(
+		context.Background(),
+		factory,
+		sink,
+		FeedConfig{},
+	)
+
+	if !errors.Is(err, openErr) {
+		t.Fatalf("runVideoAttempt() error = %v, want %v", err, openErr)
+	}
+	if factory.calls != 1 {
+		t.Errorf("factory calls = %d, want 1", factory.calls)
+	}
+	if len(sink.frames) != 0 {
+		t.Fatalf("consumed frame count = %d, want 0", len(sink.frames))
+	}
+}
+
+func TestRunVideoAttemptConsumesFrameThenReturnsReadFailure(t *testing.T) {
+	readErr := errors.New("read failed")
+	wantFrame := VideoFrame{
+		Index:   42,
+		Width:   1920,
+		Height:  1080,
+		Stride:  5120,
+		Size:    5120 * 1080,
+		Invalid: false,
+		Payload: []byte{1, 2, 3, 4},
+	}
+	reader := &fakeVideoReader{
+		frames:  []VideoFrame{wantFrame},
+		readErr: readErr,
+	}
+	sink := &fakeVideoSink{}
+
+	err := runVideoAttempt(
+		context.Background(),
+		&fakeVideoFactory{reader: reader},
+		sink,
+		FeedConfig{},
+	)
+
+	if !errors.Is(err, readErr) {
+		t.Fatalf("runVideoAttempt() error = %v, want %v", err, readErr)
+	}
+	if !reader.closed {
+		t.Fatal("reader was not closed")
+	}
+	if reader.readCalls != 2 {
+		t.Errorf("read calls = %d, want 2", reader.readCalls)
+	}
+	if len(sink.frames) != 1 {
+		t.Fatalf("consumed frame count = %d, want 1", len(sink.frames))
+	}
+	gotFrame := sink.frames[0]
+	if gotFrame.Index != wantFrame.Index ||
+		gotFrame.Width != wantFrame.Width ||
+		gotFrame.Height != wantFrame.Height ||
+		gotFrame.Stride != wantFrame.Stride ||
+		gotFrame.Size != wantFrame.Size ||
+		gotFrame.Invalid != wantFrame.Invalid {
+		t.Errorf("consumed frame metadata = %+v, want %+v", gotFrame, wantFrame)
+	}
+	if len(gotFrame.Payload) == 0 {
+		t.Fatal("consumed payload is empty")
+	}
+	if &gotFrame.Payload[0] != &wantFrame.Payload[0] {
+		t.Fatal("video payload was copied")
+	}
+}
+
+func TestRunVideoAttemptSinkFailureStopsReadingAndCloses(t *testing.T) {
+	sinkErr := errors.New("renderer unavailable")
+	reader := &fakeVideoReader{
+		frames: []VideoFrame{
+			{Index: 1, Payload: []byte{1}},
+			{Index: 2, Payload: []byte{2}},
+		},
+	}
+	sink := &fakeVideoSink{err: sinkErr}
+
+	err := runVideoAttempt(
+		context.Background(),
+		&fakeVideoFactory{reader: reader},
+		sink,
+		FeedConfig{},
+	)
+
+	if !errors.Is(err, sinkErr) {
+		t.Fatalf("runVideoAttempt() error = %v, want %v", err, sinkErr)
+	}
+	var typedErr *videoSinkError
+	if !errors.As(err, &typedErr) {
+		t.Fatalf("runVideoAttempt() error type = %T, want *videoSinkError", err)
+	}
+	if reader.readCalls != 1 {
+		t.Errorf("read calls = %d, want 1", reader.readCalls)
+	}
+	if !reader.closed {
+		t.Fatal("reader was not closed")
+	}
+}
+
+func TestRunVideoAttemptCanceledRead(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	reader := &fakeVideoReader{
+		read: func(ctx context.Context) (VideoFrame, error) {
+			cancel()
+			return VideoFrame{}, ctx.Err()
+		},
+	}
+
+	err := runVideoAttempt(
+		ctx,
+		&fakeVideoFactory{reader: reader},
+		&fakeVideoSink{},
+		FeedConfig{},
+	)
+
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("runVideoAttempt() error = %v, want context.Canceled", err)
+	}
+	if !reader.closed {
+		t.Fatal("reader was not closed")
+	}
+}
+
+func TestRunVideoAttemptJoinsReadAndCloseErrors(t *testing.T) {
+	readErr := errors.New("read failed")
+	closeErr := errors.New("close failed")
+	reader := &fakeVideoReader{
+		readErr:  readErr,
+		closeErr: closeErr,
+	}
+
+	err := runVideoAttempt(
+		context.Background(),
+		&fakeVideoFactory{reader: reader},
+		&fakeVideoSink{},
+		FeedConfig{},
+	)
+
+	if !errors.Is(err, readErr) {
+		t.Errorf("runVideoAttempt() error does not contain read error: %v", err)
+	}
+	if !errors.Is(err, closeErr) {
+		t.Errorf("runVideoAttempt() error does not contain close error: %v", err)
+	}
+}