add generic retry supervisor
This commit is contained in:
@@ -0,0 +1,54 @@
|
|||||||
|
package playback
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
type attemptFunc func(context.Context) error
|
||||||
|
type retryDecider func(error) bool
|
||||||
|
type waitFunc func(context.Context, time.Duration) error
|
||||||
|
|
||||||
|
func waitForRetry(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()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func runWithRetry(
|
||||||
|
ctx context.Context,
|
||||||
|
policy RetryPolicy,
|
||||||
|
attempt attemptFunc,
|
||||||
|
shouldRetry retryDecider,
|
||||||
|
wait waitFunc,
|
||||||
|
) error {
|
||||||
|
failedAttempts := 0
|
||||||
|
|
||||||
|
for {
|
||||||
|
err := attempt(ctx)
|
||||||
|
if err == nil {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
if ctx.Err() != nil {
|
||||||
|
return ctx.Err()
|
||||||
|
}
|
||||||
|
|
||||||
|
failedAttempts++
|
||||||
|
if !shouldRetry(err) || !policy.canRetry(failedAttempts) {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := wait(ctx, policy.retryDelay(failedAttempts)); err != nil {
|
||||||
|
if ctx.Err() != nil {
|
||||||
|
return ctx.Err()
|
||||||
|
}
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,230 @@
|
|||||||
|
package playback
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"errors"
|
||||||
|
"reflect"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
func testRetryPolicy(maxAttempts int) RetryPolicy {
|
||||||
|
return RetryPolicy{
|
||||||
|
MaxAttempts: maxAttempts,
|
||||||
|
InitialDelay: 500 * time.Millisecond,
|
||||||
|
MaxDelay: 10 * time.Second,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRunWithRetryFirstAttemptSucceeds(t *testing.T) {
|
||||||
|
attempts := 0
|
||||||
|
err := runWithRetry(
|
||||||
|
context.Background(),
|
||||||
|
testRetryPolicy(3),
|
||||||
|
func(context.Context) error {
|
||||||
|
attempts++
|
||||||
|
return nil
|
||||||
|
},
|
||||||
|
func(error) bool {
|
||||||
|
t.Fatal("shouldRetry called after successful attempt")
|
||||||
|
return false
|
||||||
|
},
|
||||||
|
func(context.Context, time.Duration) error {
|
||||||
|
t.Fatal("wait called after successful attempt")
|
||||||
|
return nil
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("runWithRetry() error = %v, want nil", err)
|
||||||
|
}
|
||||||
|
if attempts != 1 {
|
||||||
|
t.Fatalf("attempt count = %d, want 1", attempts)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRunWithRetryFailuresThenSuccess(t *testing.T) {
|
||||||
|
attemptErr := errors.New("attempt failed")
|
||||||
|
attempts := 0
|
||||||
|
var delays []time.Duration
|
||||||
|
|
||||||
|
err := runWithRetry(
|
||||||
|
context.Background(),
|
||||||
|
testRetryPolicy(3),
|
||||||
|
func(context.Context) error {
|
||||||
|
attempts++
|
||||||
|
if attempts < 3 {
|
||||||
|
return attemptErr
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
},
|
||||||
|
func(error) bool { return true },
|
||||||
|
func(_ context.Context, delay time.Duration) error {
|
||||||
|
delays = append(delays, delay)
|
||||||
|
return nil
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("runWithRetry() error = %v, want nil", err)
|
||||||
|
}
|
||||||
|
if attempts != 3 {
|
||||||
|
t.Errorf("attempt count = %d, want 3", attempts)
|
||||||
|
}
|
||||||
|
wantDelays := []time.Duration{500 * time.Millisecond, time.Second}
|
||||||
|
if !reflect.DeepEqual(delays, wantDelays) {
|
||||||
|
t.Errorf("retry delays = %v, want %v", delays, wantDelays)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRunWithRetryFiniteAttemptsExhausted(t *testing.T) {
|
||||||
|
attemptErr := errors.New("attempt failed")
|
||||||
|
attempts := 0
|
||||||
|
waits := 0
|
||||||
|
|
||||||
|
err := runWithRetry(
|
||||||
|
context.Background(),
|
||||||
|
testRetryPolicy(3),
|
||||||
|
func(context.Context) error {
|
||||||
|
attempts++
|
||||||
|
return attemptErr
|
||||||
|
},
|
||||||
|
func(error) bool { return true },
|
||||||
|
func(context.Context, time.Duration) error {
|
||||||
|
waits++
|
||||||
|
return nil
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
if !errors.Is(err, attemptErr) {
|
||||||
|
t.Fatalf("runWithRetry() error = %v, want %v", err, attemptErr)
|
||||||
|
}
|
||||||
|
if attempts != 3 {
|
||||||
|
t.Errorf("attempt count = %d, want 3", attempts)
|
||||||
|
}
|
||||||
|
if waits != 2 {
|
||||||
|
t.Errorf("wait count = %d, want 2", waits)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRunWithRetryUnlimitedEventuallySucceeds(t *testing.T) {
|
||||||
|
attemptErr := errors.New("attempt failed")
|
||||||
|
attempts := 0
|
||||||
|
|
||||||
|
err := runWithRetry(
|
||||||
|
context.Background(),
|
||||||
|
testRetryPolicy(0),
|
||||||
|
func(context.Context) error {
|
||||||
|
attempts++
|
||||||
|
if attempts < 20 {
|
||||||
|
return attemptErr
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
},
|
||||||
|
func(error) bool { return true },
|
||||||
|
func(context.Context, time.Duration) error { return nil },
|
||||||
|
)
|
||||||
|
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("runWithRetry() error = %v, want nil", err)
|
||||||
|
}
|
||||||
|
if attempts != 20 {
|
||||||
|
t.Fatalf("attempt count = %d, want 20", attempts)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRunWithRetryStopsWhenErrorIsNotRetryable(t *testing.T) {
|
||||||
|
attemptErr := errors.New("invalid configuration")
|
||||||
|
attempts := 0
|
||||||
|
|
||||||
|
err := runWithRetry(
|
||||||
|
context.Background(),
|
||||||
|
testRetryPolicy(0),
|
||||||
|
func(context.Context) error {
|
||||||
|
attempts++
|
||||||
|
return attemptErr
|
||||||
|
},
|
||||||
|
func(error) bool { return false },
|
||||||
|
func(context.Context, time.Duration) error {
|
||||||
|
t.Fatal("wait called for non-retryable error")
|
||||||
|
return nil
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
if !errors.Is(err, attemptErr) {
|
||||||
|
t.Fatalf("runWithRetry() error = %v, want %v", err, attemptErr)
|
||||||
|
}
|
||||||
|
if attempts != 1 {
|
||||||
|
t.Fatalf("attempt count = %d, want 1", attempts)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRunWithRetryReturnsCancellationFromAttempt(t *testing.T) {
|
||||||
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
|
attemptErr := errors.New("attempt failed")
|
||||||
|
|
||||||
|
err := runWithRetry(
|
||||||
|
ctx,
|
||||||
|
testRetryPolicy(0),
|
||||||
|
func(context.Context) error {
|
||||||
|
cancel()
|
||||||
|
return attemptErr
|
||||||
|
},
|
||||||
|
func(error) bool { return true },
|
||||||
|
func(context.Context, time.Duration) error {
|
||||||
|
t.Fatal("wait called after cancellation")
|
||||||
|
return nil
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
if !errors.Is(err, context.Canceled) {
|
||||||
|
t.Fatalf("runWithRetry() error = %v, want context.Canceled", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRunWithRetryReturnsCancellationDuringBackoff(t *testing.T) {
|
||||||
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
|
attemptErr := errors.New("attempt failed")
|
||||||
|
|
||||||
|
err := runWithRetry(
|
||||||
|
ctx,
|
||||||
|
testRetryPolicy(0),
|
||||||
|
func(context.Context) error { return attemptErr },
|
||||||
|
func(error) bool { return true },
|
||||||
|
func(ctx context.Context, _ time.Duration) error {
|
||||||
|
cancel()
|
||||||
|
return ctx.Err()
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
if !errors.Is(err, context.Canceled) {
|
||||||
|
t.Fatalf("runWithRetry() error = %v, want context.Canceled", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRunWithRetryReturnsWaitError(t *testing.T) {
|
||||||
|
attemptErr := errors.New("attempt failed")
|
||||||
|
waitErr := errors.New("wait failed")
|
||||||
|
|
||||||
|
err := runWithRetry(
|
||||||
|
context.Background(),
|
||||||
|
testRetryPolicy(0),
|
||||||
|
func(context.Context) error { return attemptErr },
|
||||||
|
func(error) bool { return true },
|
||||||
|
func(context.Context, time.Duration) error { return waitErr },
|
||||||
|
)
|
||||||
|
|
||||||
|
if !errors.Is(err, waitErr) {
|
||||||
|
t.Fatalf("runWithRetry() error = %v, want %v", err, waitErr)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestWaitForRetryReturnsCancellation(t *testing.T) {
|
||||||
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
|
cancel()
|
||||||
|
|
||||||
|
err := waitForRetry(ctx, time.Hour)
|
||||||
|
if !errors.Is(err, context.Canceled) {
|
||||||
|
t.Fatalf("waitForRetry() error = %v, want context.Canceled", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user