142 lines
4.3 KiB
Go
142 lines
4.3 KiB
Go
package controlplane
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"testing"
|
|
"time"
|
|
|
|
"google.golang.org/grpc/codes"
|
|
"google.golang.org/grpc/status"
|
|
)
|
|
|
|
func TestSessionSupervisorRetriesRecoverableFailuresWithCappedBackoff(t *testing.T) {
|
|
runner := &sequenceSessionLifecycle{results: []error{
|
|
ErrSnapshotStreamClosed,
|
|
errors.New("temporary control plane failure"),
|
|
status.Error(codes.PermissionDenied, "worker identity is not authorized"),
|
|
}}
|
|
sleeper := &recordingSessionSleeper{}
|
|
supervisor, err := NewSessionSupervisor(runner, ReconnectOptions{
|
|
InitialDelay: 100 * time.Millisecond,
|
|
MaxDelay: 150 * time.Millisecond,
|
|
}, SessionSupervisorRuntime{Sleeper: sleeper, Random: fixedSessionRandom(0.5)})
|
|
if err != nil {
|
|
t.Fatalf("NewSessionSupervisor(): %v", err)
|
|
}
|
|
|
|
err = supervisor.Run(context.Background())
|
|
if status.Code(err) != codes.PermissionDenied {
|
|
t.Fatalf("Run() code = %s, want PermissionDenied; error=%v", status.Code(err), err)
|
|
}
|
|
if runner.calls != 3 {
|
|
t.Fatalf("Run() calls = %d, want 3", runner.calls)
|
|
}
|
|
if got, want := sleeper.delays, []time.Duration{100 * time.Millisecond, 150 * time.Millisecond}; !equalSessionDelays(got, want) {
|
|
t.Fatalf("backoff delays = %v, want %v", got, want)
|
|
}
|
|
}
|
|
|
|
func TestSessionSupervisorStopsWhenContextCancelsDuringBackoff(t *testing.T) {
|
|
runner := &sequenceSessionLifecycle{results: []error{ErrSnapshotStreamClosed}}
|
|
sleeper := blockingSessionSleeper{started: make(chan struct{})}
|
|
supervisor, err := NewSessionSupervisor(runner, ReconnectOptions{
|
|
InitialDelay: time.Second,
|
|
MaxDelay: time.Second,
|
|
}, SessionSupervisorRuntime{Sleeper: sleeper, Random: fixedSessionRandom(0.5)})
|
|
if err != nil {
|
|
t.Fatalf("NewSessionSupervisor(): %v", err)
|
|
}
|
|
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
result := make(chan error, 1)
|
|
go func() { result <- supervisor.Run(ctx) }()
|
|
select {
|
|
case <-sleeper.started:
|
|
case <-time.After(time.Second):
|
|
t.Fatal("reconnect backoff did not start")
|
|
}
|
|
cancel()
|
|
if err := <-result; !errors.Is(err, context.Canceled) {
|
|
t.Fatalf("Run() error = %v, want context canceled", err)
|
|
}
|
|
if runner.calls != 1 {
|
|
t.Fatalf("Run() calls = %d, want 1", runner.calls)
|
|
}
|
|
}
|
|
|
|
func TestSessionSupervisorRejectsInvalidOptionsAndDoesNotRetryPermanentErrors(t *testing.T) {
|
|
if _, err := NewSessionSupervisor(&sequenceSessionLifecycle{}, ReconnectOptions{}); !errors.Is(err, ErrInvalidReconnectOptions) {
|
|
t.Fatalf("NewSessionSupervisor(invalid options) error = %v, want ErrInvalidReconnectOptions", err)
|
|
}
|
|
|
|
runner := &sequenceSessionLifecycle{results: []error{status.Error(codes.InvalidArgument, "invalid worker")}}
|
|
sleeper := &recordingSessionSleeper{}
|
|
supervisor, err := NewSessionSupervisor(runner, ReconnectOptions{
|
|
InitialDelay: time.Millisecond,
|
|
MaxDelay: time.Millisecond,
|
|
}, SessionSupervisorRuntime{Sleeper: sleeper, Random: fixedSessionRandom(0.5)})
|
|
if err != nil {
|
|
t.Fatalf("NewSessionSupervisor(): %v", err)
|
|
}
|
|
if err := supervisor.Run(context.Background()); status.Code(err) != codes.InvalidArgument {
|
|
t.Fatalf("Run() code = %s, want InvalidArgument; error=%v", status.Code(err), err)
|
|
}
|
|
if runner.calls != 1 || len(sleeper.delays) != 0 {
|
|
t.Fatalf("permanent error calls=%d delays=%v, want one call and no delay", runner.calls, sleeper.delays)
|
|
}
|
|
}
|
|
|
|
type sequenceSessionLifecycle struct {
|
|
results []error
|
|
calls int
|
|
}
|
|
|
|
func (runner *sequenceSessionLifecycle) Run(context.Context) error {
|
|
runner.calls++
|
|
if len(runner.results) == 0 {
|
|
return ErrSnapshotStreamClosed
|
|
}
|
|
result := runner.results[0]
|
|
runner.results = runner.results[1:]
|
|
return result
|
|
}
|
|
|
|
type recordingSessionSleeper struct {
|
|
delays []time.Duration
|
|
}
|
|
|
|
func (sleeper *recordingSessionSleeper) Sleep(_ context.Context, delay time.Duration) error {
|
|
sleeper.delays = append(sleeper.delays, delay)
|
|
return nil
|
|
}
|
|
|
|
type blockingSessionSleeper struct {
|
|
started chan struct{}
|
|
}
|
|
|
|
func (sleeper blockingSessionSleeper) Sleep(ctx context.Context, _ time.Duration) error {
|
|
close(sleeper.started)
|
|
<-ctx.Done()
|
|
return ctx.Err()
|
|
}
|
|
|
|
type fixedSessionRandom float64
|
|
|
|
func (random fixedSessionRandom) Float64() float64 {
|
|
return float64(random)
|
|
}
|
|
|
|
func equalSessionDelays(left, right []time.Duration) bool {
|
|
if len(left) != len(right) {
|
|
return false
|
|
}
|
|
for index := range left {
|
|
if left[index] != right[index] {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|