diff --git a/client/grpc/retry.go b/client/grpc/retry.go new file mode 100644 index 000000000..754ffa341 --- /dev/null +++ b/client/grpc/retry.go @@ -0,0 +1,49 @@ +package grpc + +import ( + "context" + "errors" + "time" + + "github.com/cenkalti/backoff/v4" + + "github.com/netbirdio/netbird/client/netstate" +) + +// Retry mirrors backoff.Retry, but the sleep between attempts also wakes on +// OS network availability transitions: an operation cut down by a network +// change retries the moment the network settles instead of sleeping through +// the recovery. A nil netState never fires, leaving plain backoff.Retry +// behavior. +func Retry(ctx context.Context, operation backoff.Operation, bo backoff.BackOff, netState *netstate.State) error { + bo.Reset() + for { + err := operation() + if err == nil { + return nil + } + + var permanent *backoff.PermanentError + if errors.As(err, &permanent) { + return permanent.Err + } + + next := bo.NextBackOff() + if next == backoff.Stop { + if cerr := ctx.Err(); cerr != nil { + return cerr + } + return err + } + + timer := time.NewTimer(next) + select { + case <-timer.C: + case <-netState.Changed(): + timer.Stop() + case <-ctx.Done(): + timer.Stop() + return ctx.Err() + } + } +} diff --git a/client/grpc/retry_test.go b/client/grpc/retry_test.go new file mode 100644 index 000000000..ea148b987 --- /dev/null +++ b/client/grpc/retry_test.go @@ -0,0 +1,91 @@ +package grpc + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/cenkalti/backoff/v4" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/netbirdio/netbird/client/netstate" +) + +func TestRetryWakesOnNetworkChange(t *testing.T) { + ns := netstate.New() + attempts := 0 + operation := func() error { + attempts++ + if attempts == 1 { + return errors.New("cut by network change") + } + return nil + } + + go func() { + time.Sleep(20 * time.Millisecond) + ns.Set(false) + }() + + start := time.Now() + err := Retry(context.Background(), operation, backoff.NewConstantBackOff(time.Minute), ns) + + require.NoError(t, err) + assert.Equal(t, 2, attempts) + assert.Less(t, time.Since(start), time.Second, "the transition must cut the minute-long sleep short") +} + +func TestRetryPermanentError(t *testing.T) { + sentinel := errors.New("permission denied") + operation := func() error { + return backoff.Permanent(sentinel) + } + + err := Retry(context.Background(), operation, backoff.NewConstantBackOff(time.Millisecond), nil) + assert.ErrorIs(t, err, sentinel) +} + +func TestRetryNilNetState(t *testing.T) { + attempts := 0 + operation := func() error { + attempts++ + if attempts < 3 { + return errors.New("transient") + } + return nil + } + + err := Retry(context.Background(), operation, backoff.NewConstantBackOff(time.Millisecond), nil) + require.NoError(t, err) + assert.Equal(t, 3, attempts) +} + +func TestRetryStops(t *testing.T) { + failure := errors.New("still failing") + operation := func() error { + return failure + } + + err := Retry(context.Background(), operation, &backoff.StopBackOff{}, nil) + assert.ErrorIs(t, err, failure) +} + +func TestRetryCtxCancelDuringSleep(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + operation := func() error { + return errors.New("failing") + } + + go func() { + time.Sleep(20 * time.Millisecond) + cancel() + }() + + start := time.Now() + err := Retry(ctx, operation, backoff.NewConstantBackOff(time.Minute), netstate.New()) + + assert.ErrorIs(t, err, context.Canceled) + assert.Less(t, time.Since(start), time.Second) +} diff --git a/client/netsweep/netsweep.go b/client/netsweep/netsweep.go index ac8cd2d05..46bc0a709 100644 --- a/client/netsweep/netsweep.go +++ b/client/netsweep/netsweep.go @@ -13,13 +13,18 @@ import ( "sync" "time" + "github.com/cenkalti/backoff/v4" log "github.com/sirupsen/logrus" + + "github.com/netbirdio/netbird/client/netstate" ) // DefaultSweepDelay absorbs network flapping while the OS settles on a // default network before the stale registrations are cut. const DefaultSweepDelay = 500 * time.Millisecond +const recentMarkWindow = 3 * time.Second + // Config customizes a Sweeper. The zero value applies the defaults. type Config struct { // SweepDelay overrides DefaultSweepDelay when positive. @@ -96,6 +101,7 @@ type Sweeper struct { gen uint64 timer *time.Timer sweepDelay time.Duration + lastMark time.Time } // New creates an empty sweeper with the default configuration. @@ -179,6 +185,7 @@ func (s *Sweeper) MarkNetworkChange() { s.mu.Lock() s.gen++ cutoff := s.gen + s.lastMark = time.Now() if s.timer != nil { s.timer.Stop() } @@ -189,6 +196,27 @@ func (s *Sweeper) MarkNetworkChange() { s.mu.Unlock() } +// QuickRetryBackoff wraps bo so that after each Reset the first retry comes +// quickly when the disconnect followed a recent network change and the +// network is online. Any other failure keeps bo's spread, so the clients of +// a restarted server still scatter their reconnects. A nil sweeper returns +// bo unchanged. +func (s *Sweeper) QuickRetryBackoff(ctx context.Context, bo backoff.BackOff, netState *netstate.State) backoff.BackOff { + if s == nil { + return bo + } + return backoff.WithContext(newQuickRetryBackoff(bo, s, netState), ctx) +} + +func (s *Sweeper) markedRecently() bool { + if s == nil { + return false + } + s.mu.Lock() + defer s.mu.Unlock() + return !s.lastMark.IsZero() && time.Since(s.lastMark) < recentMarkWindow +} + // sweep closes the registered connections and aborts the in-flight dials // older than cutoff, and returns how many connections it closed. A dial // whose connection was not yet handed to WrapConn is marked, so the late diff --git a/client/netsweep/quick_retry.go b/client/netsweep/quick_retry.go new file mode 100644 index 000000000..524a5c50c --- /dev/null +++ b/client/netsweep/quick_retry.go @@ -0,0 +1,39 @@ +package netsweep + +import ( + "time" + + "github.com/cenkalti/backoff/v4" + + "github.com/netbirdio/netbird/client/netstate" +) + +const quickRetryDelay = 200 * time.Millisecond + +type quickRetryBackoff struct { + backoff.BackOff + sweeper *Sweeper + netState *netstate.State + used bool +} + +func newQuickRetryBackoff(bo backoff.BackOff, sweeper *Sweeper, netState *netstate.State) *quickRetryBackoff { + return &quickRetryBackoff{ + BackOff: bo, + sweeper: sweeper, + netState: netState, + } +} + +func (b *quickRetryBackoff) NextBackOff() time.Duration { + if !b.used && b.sweeper.markedRecently() && b.netState.IsOnline() { + b.used = true + return quickRetryDelay + } + return b.BackOff.NextBackOff() +} + +func (b *quickRetryBackoff) Reset() { + b.used = false + b.BackOff.Reset() +} diff --git a/client/netsweep/quick_retry_test.go b/client/netsweep/quick_retry_test.go new file mode 100644 index 000000000..5505862c5 --- /dev/null +++ b/client/netsweep/quick_retry_test.go @@ -0,0 +1,58 @@ +package netsweep + +import ( + "context" + "testing" + "time" + + "github.com/cenkalti/backoff/v4" + "github.com/stretchr/testify/assert" +) + +func TestQuickRetryAfterRecentMark(t *testing.T) { + sweeper := New() + sweeper.MarkNetworkChange() + + slow := backoff.NewConstantBackOff(5 * time.Second) + bo := sweeper.QuickRetryBackoff(context.Background(), slow, nil) + + assert.Equal(t, quickRetryDelay, bo.NextBackOff(), "first retry after a mark must be quick") + assert.Equal(t, 5*time.Second, bo.NextBackOff(), "second retry must fall back to the wrapped backoff") + + bo.Reset() + assert.Equal(t, quickRetryDelay, bo.NextBackOff(), "reset must re-arm the quick retry") +} + +func TestQuickRetryWithoutMarkKeepsSpread(t *testing.T) { + sweeper := New() + + slow := backoff.NewConstantBackOff(5 * time.Second) + bo := sweeper.QuickRetryBackoff(context.Background(), slow, nil) + + assert.Equal(t, 5*time.Second, bo.NextBackOff(), "without a mark the wrapped backoff decides") + + sweeper.mu.Lock() + sweeper.lastMark = time.Now().Add(-recentMarkWindow) + sweeper.mu.Unlock() + assert.Equal(t, 5*time.Second, bo.NextBackOff(), "a stale mark must not trigger the quick retry") +} + +func TestQuickRetryNilSweeperPassthrough(t *testing.T) { + var sweeper *Sweeper + + slow := backoff.NewConstantBackOff(5 * time.Second) + bo := sweeper.QuickRetryBackoff(context.Background(), slow, nil) + + assert.Equal(t, backoff.BackOff(slow), bo, "nil sweeper must return the backoff unchanged") +} + +func TestQuickRetryHonorsContext(t *testing.T) { + sweeper := New() + sweeper.MarkNetworkChange() + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + bo := sweeper.QuickRetryBackoff(ctx, backoff.NewConstantBackOff(time.Millisecond), nil) + + assert.Equal(t, backoff.Stop, bo.NextBackOff(), "cancelled context must stop the retry loop") +} diff --git a/shared/management/client/grpc.go b/shared/management/client/grpc.go index 73eed247a..cd250b5f7 100644 --- a/shared/management/client/grpc.go +++ b/shared/management/client/grpc.go @@ -235,7 +235,7 @@ func (c *GrpcClient) withMgmtStream( ctx context.Context, handler func(ctx context.Context, serverPubKey wgtypes.Key, backOff backoff.BackOff) error, ) error { - backOff := defaultBackoff(ctx) + backOff := c.sweeper.QuickRetryBackoff(ctx, defaultBackoff(ctx), c.netState) operation := func() error { // suspend reconnect attempts while the OS reports no usable network. // Wait only errors on a cancelled context, which means shutdown, so @@ -247,14 +247,21 @@ func (c *GrpcClient) withMgmtStream( backOff.Reset() } - log.Debugf("management connection state %v", c.conn.GetState()) connState := c.conn.GetState() - + log.Debugf("management connection state %v", connState) if connState == connectivity.Shutdown { return backoff.Permanent(fmt.Errorf("connection to management has been shut down")) - } else if !(connState == connectivity.Ready || connState == connectivity.Idle) { + } + if !(connState == connectivity.Ready || connState == connectivity.Idle) { + // A dial may already be in flight (e.g. the other stream triggered + // it after a network change); wait for it to settle and proceed if + // the channel became usable, instead of burning a backoff round on + // a successful dial. A failed dial errors out as before. c.conn.WaitForStateChange(ctx, connState) - return fmt.Errorf("connection to management is not ready and in %s state", connState) + connState = c.conn.GetState() + if !(connState == connectivity.Ready || connState == connectivity.Idle) { + return fmt.Errorf("connection to management is not ready and in %s state", connState) + } } serverPubKey, err := c.getServerPublicKey() @@ -266,7 +273,7 @@ func (c *GrpcClient) withMgmtStream( return handler(ctx, *serverPubKey, backOff) } - err := backoff.Retry(operation, backOff) + err := nbgrpc.Retry(ctx, operation, backOff, c.netState) if err != nil { log.Warnf("exiting the Management service connection retry loop due to the unrecoverable error: %s", err) } diff --git a/shared/signal/client/grpc.go b/shared/signal/client/grpc.go index bb0d578b7..73c482e8f 100644 --- a/shared/signal/client/grpc.go +++ b/shared/signal/client/grpc.go @@ -198,7 +198,7 @@ func defaultBackoff(ctx context.Context) backoff.BackOff { // The connection retry logic will try to reconnect for 30 min and if wasn't successful will propagate the error to the function caller. func (c *GrpcClient) Receive(ctx context.Context, msgHandler func(msg *proto.Message) error) error { - var backOff = defaultBackoff(ctx) + backOff := c.sweeper.QuickRetryBackoff(ctx, defaultBackoff(ctx), c.netState) operation := func() error { // suspend reconnect attempts while the OS reports no usable network. @@ -213,13 +213,21 @@ func (c *GrpcClient) Receive(ctx context.Context, msgHandler func(msg *proto.Mes c.notifyStreamDisconnected() - log.Debugf("signal connection state %v", c.signalConn.GetState()) connState := c.signalConn.GetState() + log.Debugf("signal connection state %v", connState) if connState == connectivity.Shutdown { return backoff.Permanent(fmt.Errorf("connection to signal has been shut down")) - } else if !(connState == connectivity.Ready || connState == connectivity.Idle) { + } + if !(connState == connectivity.Ready || connState == connectivity.Idle) { + // A dial may already be in flight (e.g. triggered by another RPC + // after a network change); wait for it to settle and proceed if + // the channel became usable, instead of burning a backoff round on + // a successful dial. A failed dial errors out as before. c.signalConn.WaitForStateChange(ctx, connState) - return fmt.Errorf("connection to signal is not ready and in %s state", connState) + connState = c.signalConn.GetState() + if !(connState == connectivity.Ready || connState == connectivity.Idle) { + return fmt.Errorf("connection to signal is not ready and in %s state", connState) + } } // connect to Signal stream identifying ourselves with a public WireGuard key @@ -273,7 +281,7 @@ func (c *GrpcClient) Receive(ctx context.Context, msgHandler func(msg *proto.Mes return nil } - err := backoff.Retry(operation, backOff) + err := nbgrpc.Retry(ctx, operation, backOff, c.netState) if err != nil { log.Errorf("exiting the Signal service connection retry loop due to the unrecoverable error: %v", err) return err