define playback retry rules and states

cmd/mxl-player/main.go

@@ -243,7 +243,7 @@ func main() {
 
 	switch {
 	case args.VideoFlowId != "" && args.AudioFlowId != "":
-		syncSrc, err = source.OpenSync(args.Domain, args.VideoFlowId, args.AudioFlowId)
+		syncSrc, err = source.OpenSameDomainSync(args.Domain, args.VideoFlowId, args.AudioFlowId)
 		if err != nil {
 			log.Fatalf("sync source: %v", err)
 		}
@@ -378,7 +378,7 @@ func main() {
 		// Try once. Return error if fails — caller loops back to select
 		// and can pick up new reconnect params or a new grant.
 		if params.video != "" && params.audio != "" {
-			s, e := source.OpenSync(params.domain, params.video, params.audio)
+			s, e := source.OpenSameDomainSync(params.domain, params.video, params.audio)
 			if e == nil {
 				if r != nil {
 					newSize := vk.DeviceSize(s.Stride()) * vk.DeviceSize(s.Height())

internal/playback/retry.go

@@ -0,0 +1,23 @@
+package playback
+
+import "time"
+
+func (p RetryPolicy) canRetry(failedAttempts int) bool {
+	return p.MaxAttempts == 0 || failedAttempts < p.MaxAttempts
+}
+
+func (p RetryPolicy) retryDelay(failedAttempts int) time.Duration {
+	delay := p.InitialDelay
+
+	for attempt := 1; attempt < failedAttempts; attempt++ {
+		if delay >= p.MaxDelay/2 {
+			return p.MaxDelay
+		}
+		delay *= 2
+	}
+
+	if delay > p.MaxDelay {
+		return p.MaxDelay
+	}
+	return delay
+}

internal/playback/retry_test.go

@@ -0,0 +1,91 @@
+package playback
+
+import (
+	"testing"
+	"time"
+)
+
+func TestFiniteAttempts(t *testing.T) {
+	rp := RetryPolicy{
+		MaxAttempts:  3,
+		InitialDelay: 500 * time.Millisecond,
+		MaxDelay:     10 * time.Second,
+	}
+	if !rp.canRetry(1) {
+		t.Fatal("MaxAttempts=3, failed=1, but can't retry")
+	}
+	if !rp.canRetry(2) {
+		t.Fatal("MaxAttempts=3, failed=2, but can't retry")
+	}
+	if rp.canRetry(3) {
+		t.Fatal("MaxAttempts=3, failed=3, but can retry")
+	}
+}
+
+func TestOneAllowedAttempt(t *testing.T) {
+	rp := RetryPolicy{
+		MaxAttempts:  1,
+		InitialDelay: 500 * time.Millisecond,
+		MaxDelay:     10 * time.Second,
+	}
+	if rp.canRetry(1) {
+		t.Fatal("MaxAttempts=1, failed=1, but can retry")
+	}
+}
+
+func TestUnlimitedAttempts(t *testing.T) {
+	rp := RetryPolicy{
+		MaxAttempts:  0,
+		InitialDelay: 500 * time.Millisecond,
+		MaxDelay:     10 * time.Second,
+	}
+	for failedAttempts := 1; failedAttempts <= 10; failedAttempts++ {
+		if !rp.canRetry(failedAttempts) {
+			t.Fatalf(
+				"canRetry(%d) = false for unlimited policy",
+				failedAttempts,
+			)
+		}
+	}
+}
+
+func TestBackoff(t *testing.T) {
+	rp := RetryPolicy{
+		MaxAttempts:  0,
+		InitialDelay: 500 * time.Millisecond,
+		MaxDelay:     10 * time.Second,
+	}
+	want := []time.Duration{
+		500 * time.Millisecond,
+		1 * time.Second,
+		2 * time.Second,
+		4 * time.Second,
+		8 * time.Second,
+		10 * time.Second,
+		10 * time.Second,
+	}
+	for i, wantDelay := range want {
+		failedAttempts := i + 1
+		got := rp.retryDelay(failedAttempts)
+		if got != wantDelay {
+			t.Errorf(
+				"retryDelay(%d) = %s, want %s",
+				failedAttempts,
+				got,
+				wantDelay,
+			)
+		}
+	}
+}
+
+func TestRetryDelayLargeFailureCount(t *testing.T) {
+	policy := RetryPolicy{
+		MaxAttempts:  0,
+		InitialDelay: 500 * time.Millisecond,
+		MaxDelay:     10 * time.Second,
+	}
+
+	if got := policy.retryDelay(1_000_000); got != policy.MaxDelay {
+		t.Fatalf("retryDelay() = %s, want cap %s", got, policy.MaxDelay)
+	}
+}

internal/playback/state.go

@@ -0,0 +1,12 @@
+package playback
+
+type State uint8
+
+const (
+	StateIdle State = iota
+	StateConnecting
+	StatePlaying
+	StateReconnecting
+	StateFailed
+	StateStopping
+)

internal/source/errors.go

@@ -0,0 +1,53 @@
+package source
+
+import (
+	"errors"
+	"fmt"
+)
+
+type ErrorKind uint8
+
+const (
+	ErrorKindUnknown       ErrorKind = iota
+	ErrorKindTemporary               // timeout or temporarily early/late data i.e. wait or resync
+	ErrorKindUnavailable             // producer/flow disappeared
+	ErrorKindInvalidConfig           // wrong media type, invalid rate, etc.
+)
+
+type SourceError struct {
+	Op   string
+	Kind ErrorKind
+	Err  error
+}
+
+func (e *SourceError) Error() string {
+	if e.Op == "" {
+		return e.Err.Error()
+	}
+	return fmt.Sprintf("%s: %v", e.Op, e.Err)
+}
+
+func (e *SourceError) Unwrap() error {
+	return e.Err
+}
+
+// KindOf returns the source error category contained in err.
+// It returns ErrorKindUnknown when err has no SourceError in its chain.
+func KindOf(err error) ErrorKind {
+	var sourceErr *SourceError
+	if errors.As(err, &sourceErr) {
+		return sourceErr.Kind
+	}
+	return ErrorKindUnknown
+}
+
+func wrapError(op string, kind ErrorKind, err error) error {
+	if err == nil {
+		return nil
+	}
+	return &SourceError{
+		Op:   op,
+		Kind: kind,
+		Err:  err,
+	}
+}

internal/source/errors_test.go

@@ -0,0 +1,54 @@
+package source
+
+import (
+	"errors"
+	"fmt"
+	"testing"
+)
+
+func TestKindOfThroughWrapping(t *testing.T) {
+	base := errors.New("producer disappeared")
+	wrapped := wrapError("read video", ErrorKindUnavailable, base)
+	outer := fmt.Errorf("worker failed: %w", wrapped)
+
+	if got := KindOf(outer); got != ErrorKindUnavailable {
+		t.Fatalf("KindOf() = %v, want %v", got, ErrorKindUnavailable)
+	}
+	if !errors.Is(outer, base) {
+		t.Fatal("wrapped error does not preserve its cause")
+	}
+	if wrapError("nil", 0, nil) != nil {
+		t.Fatal("nil error is not wrapped as nil")
+	}
+}
+
+func TestKindOfUnknown(t *testing.T) {
+	if got := KindOf(errors.New("ordinary error")); got != ErrorKindUnknown {
+		t.Fatalf("KindOf() = %v, want %v", got, ErrorKindUnknown)
+	}
+}
+
+func TestSourceErrorWithOperation(t *testing.T) {
+	err := &SourceError{
+		Op:   "read video",
+		Kind: ErrorKindTemporary,
+		Err:  errors.New("timeout"),
+	}
+
+	const want = "read video: timeout"
+	if got := err.Error(); got != want {
+		t.Fatalf("Error() = %q, want %q", got, want)
+	}
+}
+
+func TestSourceErrorWithoutOperation(t *testing.T) {
+	err := &SourceError{
+		Kind: ErrorKindTemporary,
+		Err:  errors.New("timeout"),
+	}
+
+	const want = "timeout"
+	if got := err.Error(); got != want {
+		t.Fatalf("Error() = %q, want %q", got, want)
+	}
+}

internal/source/source.go

@@ -46,42 +46,59 @@ type Source struct {
 func Open(domain, flowID string) (*Source, error) {
 	inst, err := mxl.NewInstance(domain, "")
 	if err != nil {
-		return nil, fmt.Errorf("NewInstance: %w", err)
+		return nil, wrapError("new MXL instance", ErrorKindUnavailable, err)
 	}
 	r, err := inst.NewReader(flowID)
 	if err != nil {
 		inst.Close()
-		return nil, fmt.Errorf("NewReader: %w", err)
+		return nil, wrapError("open video reader", ErrorKindUnavailable, err)
 	}
 	info, err := r.Info()
 	if err != nil {
 		r.Close()
 		inst.Close()
-		return nil, fmt.Errorf("Info: %w", err)
+		return nil, wrapError("get video info", ErrorKindUnavailable, err)
 	}
 	def, err := inst.FlowDef(flowID)
 	if err != nil {
 		r.Close()
 		inst.Close()
-		return nil, fmt.Errorf("FlowDef: %w", err)
+		return nil, wrapError("read video flow definition", ErrorKindUnavailable, err)
 	}
 	var fd flowDef
 	if err := json.Unmarshal([]byte(def), &fd); err != nil {
 		r.Close()
 		inst.Close()
-		return nil, fmt.Errorf("parse flow def: %w", err)
+		return nil, wrapError("parse video flow definition JSON", ErrorKindInvalidConfig, err)
 	}
 	if fd.FrameWidth == 0 || fd.FrameHeight == 0 {
 		r.Close()
 		inst.Close()
-		return nil, fmt.Errorf("flow has no video dimensions (not a video flow?)")
+		return nil, wrapError(
+			"validate video flow",
+			ErrorKindInvalidConfig,
+			errors.New("flow has no video dimensions"),
+		)
 	}
 	rate := info.Config.Common.GrainRate
 	idx := mxl.CurrentIndex(rate)
 	if idx == mxl.UndefinedIndex {
 		r.Close()
 		inst.Close()
-		return nil, fmt.Errorf("invalid grain rate: %d/%d", rate.Num, rate.Den)
+		return nil, wrapError(
+			"validate video flow",
+			ErrorKindInvalidConfig,
+			fmt.Errorf("invalid grain rate: %d/%d", rate.Num, rate.Den),
+		)
+	}
+	if len(info.Config.Discrete.SliceSizes) == 0 {
+		r.Close()
+		inst.Close()
+		return nil, wrapError(
+			"validate video flow",
+			ErrorKindInvalidConfig,
+			errors.New("video flow has no slice sizes"),
+		)
 	}
 	return &Source{
 		inst:   inst,
@@ -101,10 +118,6 @@ func (s *Source) Close() error {
 	return s.inst.Close()
 }
 
-func (s *Source) Next(timeout time.Duration) (Frame, error) {
-	return s.NextCtx(context.Background(), timeout)
-}
-
 func (s *Source) NextCtx(ctx context.Context, timeout time.Duration) (Frame, error) {
 	for {
 		select {
@@ -165,36 +178,64 @@ type AudioFrame struct {
 func OpenAudio(domain, flowID string) (*AudioSource, error) {
 	inst, err := mxl.NewInstance(domain, "")
 	if err != nil {
-		return nil, fmt.Errorf("NewInstance: %w", err)
+		return nil, wrapError("new MXL instance", ErrorKindUnavailable, err)
 	}
 	r, err := inst.NewReader(flowID)
 	if err != nil {
 		inst.Close()
-		return nil, fmt.Errorf("NewReader: %w", err)
+		return nil, wrapError("open audio reader", ErrorKindUnavailable, err)
 	}
 	info, err := r.Info()
 	if err != nil {
 		r.Close()
 		inst.Close()
-		return nil, fmt.Errorf("Info: %w", err)
+		return nil, wrapError("get audio info", ErrorKindUnavailable, err)
 	}
 	if info.Config.Common.Format.IsDiscrete() {
 		r.Close()
 		inst.Close()
-		return nil, fmt.Errorf("flow is discrete (not audio)")
+		return nil, wrapError(
+			"validate audio flow",
+			ErrorKindInvalidConfig,
+			errors.New("audio flow is discrete"),
+		)
+	}
+	channels := uint64(info.Config.Continuous.ChannelCount)
+	if channels == 0 {
+		r.Close()
+		inst.Close()
+		return nil, wrapError(
+			"validate audio flow",
+			ErrorKindInvalidConfig,
+			errors.New("audio flow has no channels"),
+		)
+	}
+	rate := info.Config.Common.GrainRate
+	if rate.Num <= 0 || rate.Den <= 0 {
+		r.Close()
+		inst.Close()
+		return nil, wrapError(
+			"validate audio flow",
+			ErrorKindInvalidConfig,
+			fmt.Errorf("invalid audio rate: %d/%d", rate.Num, rate.Den),
+		)
 	}
 	idx := info.Runtime.HeadIndex
 	if idx == 0 {
 		r.Close()
 		inst.Close()
-		return nil, fmt.Errorf("flow has no head yet (no producer?)")
+		return nil, wrapError(
+			"open audio reader",
+			ErrorKindUnavailable,
+			errors.New("audio flow has no producer data"),
+		)
 	}
 	return &AudioSource{
 		inst:  inst,
 		r:     r,
 		info:  info,
-		rate:  info.Config.Common.GrainRate,
-		chans: uint64(info.Config.Continuous.ChannelCount),
+		rate:  rate,
+		chans: channels,
 		idx:   idx,
 	}, nil
 }
@@ -264,22 +305,25 @@ type SyncSource struct {
 	width, height, stride uint32
 }
 
-func OpenSync(domain, videoFlow, audioFlow string) (*SyncSource, error) {
+// OpenSameDomainSync opens a native MXL synchronization group.
+// Both feeds must belong to the supplied domain because go-mxl sync groups
+// cannot contain readers from different MXL instances.
+func OpenSameDomainSync(domain, videoFlow, audioFlow string) (*SyncSource, error) {
 	inst, err := mxl.NewInstance(domain, "")
 	if err != nil {
-		return nil, fmt.Errorf("NewInstance: %w", err)
+		return nil, wrapError("new MXL instance", ErrorKindUnavailable, err)
 	}
 
 	vr, err := inst.NewReader(videoFlow)
 	if err != nil {
 		inst.Close()
-		return nil, fmt.Errorf("NewReader(video): %w", err)
+		return nil, wrapError("open video reader", ErrorKindUnavailable, err)
 	}
 	ar, err := inst.NewReader(audioFlow)
 	if err != nil {
 		vr.Close()
 		inst.Close()
-		return nil, fmt.Errorf("NewReader(audio): %w", err)
+		return nil, wrapError("open audio reader", ErrorKindUnavailable, err)
 	}
 
 	vInfo, err := vr.Info()
@@ -287,26 +331,56 @@ func OpenSync(domain, videoFlow, audioFlow string) (*SyncSource, error) {
 		ar.Close()
 		vr.Close()
 		inst.Close()
-		return nil, fmt.Errorf("Info(video): %w", err)
+		return nil, wrapError("get video info", ErrorKindUnavailable, err)
 	}
 	if !vInfo.Config.Common.Format.IsDiscrete() {
 		ar.Close()
 		vr.Close()
 		inst.Close()
-		return nil, fmt.Errorf("video flow is not discrete")
+		return nil, wrapError(
+			"validate video flow",
+			ErrorKindInvalidConfig,
+			errors.New("video flow is continuous"),
+		)
 	}
 	aInfo, err := ar.Info()
 	if err != nil {
 		ar.Close()
 		vr.Close()
 		inst.Close()
-		return nil, fmt.Errorf("Info(audio): %w", err)
+		return nil, wrapError("get audio info", ErrorKindUnavailable, err)
 	}
 	if aInfo.Config.Common.Format.IsDiscrete() {
 		ar.Close()
 		vr.Close()
 		inst.Close()
-		return nil, fmt.Errorf("audio flow is not continuous")
+		return nil, wrapError(
+			"validate audio flow",
+			ErrorKindInvalidConfig,
+			errors.New("audio flow is discrete"),
+		)
+	}
+	channels := uint64(aInfo.Config.Continuous.ChannelCount)
+	if channels == 0 {
+		ar.Close()
+		vr.Close()
+		inst.Close()
+		return nil, wrapError(
+			"validate audio flow",
+			ErrorKindInvalidConfig,
+			errors.New("audio flow has no channels"),
+		)
+	}
+	aRate := aInfo.Config.Common.GrainRate
+	if aRate.Num <= 0 || aRate.Den <= 0 {
+		ar.Close()
+		vr.Close()
+		inst.Close()
+		return nil, wrapError(
+			"validate audio flow",
+			ErrorKindInvalidConfig,
+			fmt.Errorf("invalid audio rate: %d/%d", aRate.Num, aRate.Den),
+		)
 	}
 
 	def, err := inst.FlowDef(videoFlow)
@@ -314,14 +388,38 @@ func OpenSync(domain, videoFlow, audioFlow string) (*SyncSource, error) {
 		ar.Close()
 		vr.Close()
 		inst.Close()
-		return nil, fmt.Errorf("FlowDef: %w", err)
+		return nil, wrapError("read video flow definition", ErrorKindUnavailable, err)
 	}
 	var fd flowDef
 	if err := json.Unmarshal([]byte(def), &fd); err != nil {
 		ar.Close()
 		vr.Close()
 		inst.Close()
-		return nil, fmt.Errorf("parse flow def: %w", err)
+		return nil, wrapError("parse video flow definition JSON", ErrorKindInvalidConfig, err)
+	}
+	if fd.FrameWidth <= 0 || fd.FrameHeight <= 0 {
+		ar.Close()
+		vr.Close()
+		inst.Close()
+		return nil, wrapError(
+			"validate video flow",
+			ErrorKindInvalidConfig,
+			fmt.Errorf(
+				"invalid video dimensions: %dx%d",
+				fd.FrameWidth,
+				fd.FrameHeight,
+			),
+		)
+	}
+	if len(vInfo.Config.Discrete.SliceSizes) == 0 {
+		ar.Close()
+		vr.Close()
+		inst.Close()
+		return nil, wrapError(
+			"validate video flow",
+			ErrorKindInvalidConfig,
+			errors.New("video flow has no slice sizes"),
+		)
 	}
 
 	vRate := vInfo.Config.Common.GrainRate
@@ -330,7 +428,11 @@ func OpenSync(domain, videoFlow, audioFlow string) (*SyncSource, error) {
 		ar.Close()
 		vr.Close()
 		inst.Close()
-		return nil, fmt.Errorf("invalid video rate %d/%d", vRate.Num, vRate.Den)
+		return nil, wrapError(
+			"validate video flow",
+			ErrorKindInvalidConfig,
+			fmt.Errorf("invalid grain rate: %d/%d", vRate.Num, vRate.Den),
+		)
 	}
 
 	group, err := inst.NewSyncGroup()
@@ -338,21 +440,33 @@ func OpenSync(domain, videoFlow, audioFlow string) (*SyncSource, error) {
 		ar.Close()
 		vr.Close()
 		inst.Close()
-		return nil, fmt.Errorf("NewSyncGroup: %w", err)
+		return nil, wrapError(
+			"create native sync group",
+			ErrorKindUnavailable,
+			err,
+		)
 	}
 	if err := group.AddReader(vr); err != nil {
 		group.Close()
 		ar.Close()
 		vr.Close()
 		inst.Close()
-		return nil, fmt.Errorf("AddReader(video): %w", err)
+		return nil, wrapError(
+			"add video reader to native sync group",
+			ErrorKindUnavailable,
+			err,
+		)
 	}
 	if err := group.AddReader(ar); err != nil {
 		group.Close()
 		ar.Close()
 		vr.Close()
 		inst.Close()
-		return nil, fmt.Errorf("AddReader(audio): %w", err)
+		return nil, wrapError(
+			"add audio reader to native sync group",
+			ErrorKindUnavailable,
+			err,
+		)
 	}
 
 	return &SyncSource{
@@ -361,8 +475,8 @@ func OpenSync(domain, videoFlow, audioFlow string) (*SyncSource, error) {
 		ar:     ar,
 		group:  group,
 		rate:   vRate,
-		aRate:  aInfo.Config.Common.GrainRate,
-		chans:  uint64(aInfo.Config.Continuous.ChannelCount),
+		aRate:  aRate,
+		chans:  channels,
 		idx:    idx,
 		width:  uint32(fd.FrameWidth),
 		height: uint32(fd.FrameHeight),

internal/test.go

@@ -1,7 +0,0 @@
-package main
-
-import "mxl-player/internal/imgui"
-
-func main() {
-	imgui.New()
-}