mirror of
https://github.com/netbirdio/netbird.git
synced 2026-08-19 22:21:29 +02:00
Compare commits
1 Commits
fix/androi
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
9efa3c6579 |
@@ -26,7 +26,8 @@ import (
|
||||
"github.com/netbirdio/netbird/client/internal/routemanager"
|
||||
"github.com/netbirdio/netbird/client/internal/stdnet"
|
||||
"github.com/netbirdio/netbird/client/net"
|
||||
"github.com/netbirdio/netbird/client/netevents"
|
||||
"github.com/netbirdio/netbird/client/netstate"
|
||||
"github.com/netbirdio/netbird/client/netsweep"
|
||||
"github.com/netbirdio/netbird/client/system"
|
||||
"github.com/netbirdio/netbird/formatter"
|
||||
"github.com/netbirdio/netbird/route"
|
||||
@@ -81,10 +82,13 @@ type Client struct {
|
||||
deviceName string
|
||||
uiVersion string
|
||||
networkChangeListener listener.NetworkChangeListener
|
||||
// netMgr outlives engine restarts: it mirrors the OS connectivity, not
|
||||
// the engine lifecycle. Run and RunWithoutLogin inject its state and
|
||||
// sweeper into each new ConnectClient.
|
||||
netMgr *netevents.Manager
|
||||
// netState outlives engine restarts: it mirrors the OS connectivity, not
|
||||
// the engine lifecycle. Run and RunWithoutLogin inject it into each new
|
||||
// ConnectClient, which distributes it to every reconnection loop.
|
||||
netState *netstate.State
|
||||
|
||||
// sweeper also outlives engine restarts; NotifyNetworkChange sweeps it.
|
||||
sweeper *netsweep.Sweeper
|
||||
|
||||
stateMu sync.RWMutex
|
||||
connectClient *internal.ConnectClient
|
||||
@@ -149,16 +153,16 @@ func NewClient(androidSDKVersion int, deviceName string, uiVersion string, tunAd
|
||||
|
||||
net.SetAndroidProtectSocketFn(tunAdapter.ProtectSocket)
|
||||
system.SetIFaceDiscover(iFaceDiscover)
|
||||
recorder := peer.NewRecorder("")
|
||||
return &Client{
|
||||
deviceName: deviceName,
|
||||
uiVersion: uiVersion,
|
||||
tunAdapter: tunAdapter,
|
||||
iFaceDiscover: iFaceDiscover,
|
||||
recorder: recorder,
|
||||
recorder: peer.NewRecorder(""),
|
||||
ctxCancelLock: &sync.Mutex{},
|
||||
networkChangeListener: networkChangeListener,
|
||||
netMgr: netevents.NewManager(recorder),
|
||||
netState: netstate.New(),
|
||||
sweeper: netsweep.New(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -199,9 +203,8 @@ func (c *Client) Run(platformFiles PlatformFiles, urlOpener URLOpener, isAndroid
|
||||
}
|
||||
// todo do not throw error in case of cancelled context
|
||||
ctx = internal.CtxInitState(ctx)
|
||||
|
||||
connectClient := internal.NewConnectClient(ctx, cfg, c.recorder,
|
||||
internal.WithNetEvents(c.netMgr))
|
||||
internal.WithNetworkState(c.netState), internal.WithSweeper(c.sweeper))
|
||||
c.setState(cfg, cacheDir, cfgFile, connectClient)
|
||||
// This path runs the interactive SSO flow, so reaching here means the peer
|
||||
// is authenticated again — release the latch Status() reports from. Clear
|
||||
@@ -243,7 +246,7 @@ func (c *Client) RunWithoutLogin(platformFiles PlatformFiles, dns *DNSList, dnsR
|
||||
// todo do not throw error in case of cancelled context
|
||||
ctx = internal.CtxInitState(ctx)
|
||||
connectClient := internal.NewConnectClient(ctx, cfg, c.recorder,
|
||||
internal.WithNetEvents(c.netMgr))
|
||||
internal.WithNetworkState(c.netState), internal.WithSweeper(c.sweeper))
|
||||
c.setState(cfg, cacheDir, cfgFile, connectClient)
|
||||
return connectClient.RunOnAndroid(c.tunAdapter, c.iFaceDiscover, c.networkChangeListener, slices.Clone(dns.items), dnsReadyListener, stateFile, cacheDir)
|
||||
}
|
||||
@@ -295,12 +298,9 @@ func (c *Client) GetTunSettings() (*TunSettings, error) {
|
||||
// While unavailable, the internal reconnect loops suspend their attempts and
|
||||
// the connection listener reports NoNetwork instead of Connecting; when
|
||||
// availability returns, the loops resume immediately with a fresh backoff.
|
||||
// Losing the last network also sweeps the registered connections: nothing can
|
||||
// redial while offline, so the stale sockets would otherwise stay silently
|
||||
// "connected" until their own timeouts and the client would keep reporting
|
||||
// Connected with no network at all.
|
||||
func (c *Client) SetNetworkAvailable(available bool) {
|
||||
c.netMgr.SetNetworkAvailable(available)
|
||||
c.netState.Set(available)
|
||||
c.recorder.SetNetworkAvailable(available)
|
||||
}
|
||||
|
||||
// NotifyNetworkChange marks the management, signal and relay connections
|
||||
@@ -308,7 +308,8 @@ func (c *Client) SetNetworkAvailable(available bool) {
|
||||
// whatever has not redialed on the new network by then. The engine and the
|
||||
// TUN device stay untouched.
|
||||
func (c *Client) NotifyNetworkChange() {
|
||||
c.netMgr.NotifyNetworkChange()
|
||||
c.sweeper.MarkNetworkChange()
|
||||
log.Infof("network change: connections marked stale")
|
||||
}
|
||||
|
||||
// DebugBundle generates a debug bundle, uploads it, and returns the upload key.
|
||||
|
||||
@@ -16,14 +16,9 @@ import (
|
||||
"google.golang.org/grpc"
|
||||
|
||||
nbnet "github.com/netbirdio/netbird/client/net"
|
||||
"github.com/netbirdio/netbird/client/netevents/sweep"
|
||||
"github.com/netbirdio/netbird/client/netsweep"
|
||||
)
|
||||
|
||||
// Sweeper registers in-flight dials for the network change sweep.
|
||||
type Sweeper interface {
|
||||
StartDial(ctx context.Context) *sweep.Dial
|
||||
}
|
||||
|
||||
func WithCustomDialer(_ bool, _ string) grpc.DialOption {
|
||||
return grpc.WithContextDialer(dialContext)
|
||||
}
|
||||
@@ -31,7 +26,7 @@ func WithCustomDialer(_ bool, _ string) grpc.DialOption {
|
||||
// WithSweeper dials like WithCustomDialer but registers connections and
|
||||
// dials with the sweeper. Append it after WithCustomDialer: gRPC applies
|
||||
// dial options in order, so the later context dialer wins.
|
||||
func WithSweeper(sweeper Sweeper) grpc.DialOption {
|
||||
func WithSweeper(sweeper *netsweep.Sweeper) grpc.DialOption {
|
||||
return grpc.WithContextDialer(func(ctx context.Context, addr string) (net.Conn, error) {
|
||||
dial := sweeper.StartDial(ctx)
|
||||
defer dial.Release()
|
||||
|
||||
@@ -1,19 +1,12 @@
|
||||
package grpc
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"google.golang.org/grpc"
|
||||
|
||||
"github.com/netbirdio/netbird/client/netevents/sweep"
|
||||
"github.com/netbirdio/netbird/client/netsweep"
|
||||
"github.com/netbirdio/netbird/util/wsproxy/client"
|
||||
)
|
||||
|
||||
// Sweeper registers in-flight dials for the network change sweep.
|
||||
type Sweeper interface {
|
||||
StartDial(ctx context.Context) *sweep.Dial
|
||||
}
|
||||
|
||||
// WithCustomDialer returns a gRPC dial option that uses WebSocket transport for WASM/JS environments.
|
||||
// The component parameter specifies the WebSocket proxy component path (e.g., "/management", "/signal").
|
||||
func WithCustomDialer(tlsEnabled bool, component string) grpc.DialOption {
|
||||
@@ -21,6 +14,6 @@ func WithCustomDialer(tlsEnabled bool, component string) grpc.DialOption {
|
||||
}
|
||||
|
||||
// WithSweeper is a no-op on WASM/JS: there is no network change signal.
|
||||
func WithSweeper(_ Sweeper) grpc.DialOption {
|
||||
func WithSweeper(_ *netsweep.Sweeper) grpc.DialOption {
|
||||
return grpc.EmptyDialOption{}
|
||||
}
|
||||
|
||||
@@ -6,19 +6,16 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/cenkalti/backoff/v4"
|
||||
)
|
||||
|
||||
// ChangeWatcher exposes OS network availability transitions.
|
||||
type ChangeWatcher interface {
|
||||
Changed() <-chan struct{}
|
||||
}
|
||||
"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 watcher never fires, leaving plain backoff.Retry
|
||||
// the recovery. A nil netState never fires, leaving plain backoff.Retry
|
||||
// behavior.
|
||||
func Retry(ctx context.Context, operation backoff.Operation, bo backoff.BackOff, watcher ChangeWatcher) error {
|
||||
func Retry(ctx context.Context, operation backoff.Operation, bo backoff.BackOff, netState *netstate.State) error {
|
||||
bo.Reset()
|
||||
for {
|
||||
err := operation()
|
||||
@@ -39,14 +36,10 @@ func Retry(ctx context.Context, operation backoff.Operation, bo backoff.BackOff,
|
||||
return err
|
||||
}
|
||||
|
||||
var changed <-chan struct{}
|
||||
if watcher != nil {
|
||||
changed = watcher.Changed()
|
||||
}
|
||||
timer := time.NewTimer(next)
|
||||
select {
|
||||
case <-timer.C:
|
||||
case <-changed:
|
||||
case <-netState.Changed():
|
||||
timer.Stop()
|
||||
case <-ctx.Done():
|
||||
timer.Stop()
|
||||
|
||||
@@ -10,7 +10,7 @@ import (
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/netbirdio/netbird/client/netevents/netstate"
|
||||
"github.com/netbirdio/netbird/client/netstate"
|
||||
)
|
||||
|
||||
func TestRetryWakesOnNetworkChange(t *testing.T) {
|
||||
|
||||
@@ -38,7 +38,8 @@ import (
|
||||
"github.com/netbirdio/netbird/client/internal/updater"
|
||||
"github.com/netbirdio/netbird/client/internal/updater/installer"
|
||||
nbnet "github.com/netbirdio/netbird/client/net"
|
||||
"github.com/netbirdio/netbird/client/netevents"
|
||||
"github.com/netbirdio/netbird/client/netstate"
|
||||
"github.com/netbirdio/netbird/client/netsweep"
|
||||
cProto "github.com/netbirdio/netbird/client/proto"
|
||||
"github.com/netbirdio/netbird/client/ssh"
|
||||
sshconfig "github.com/netbirdio/netbird/client/ssh/config"
|
||||
@@ -72,17 +73,28 @@ type ConnectClient struct {
|
||||
|
||||
persistSyncResponse bool
|
||||
|
||||
// netEvents gates every reconnection loop on OS-reported network
|
||||
// availability and sweeps connections on network change.
|
||||
netEvents *netevents.Manager
|
||||
// netState gates every reconnection loop on OS-reported network
|
||||
// availability. Nil (the default) disables gating; mobile platforms
|
||||
// inject it via WithNetworkState.
|
||||
netState *netstate.State
|
||||
|
||||
// sweeper cuts the management, signal and relay connections on network
|
||||
// change; nil disables it.
|
||||
sweeper *netsweep.Sweeper
|
||||
}
|
||||
|
||||
// ConnectClientOption configures optional ConnectClient behavior.
|
||||
type ConnectClientOption func(*ConnectClient)
|
||||
|
||||
// WithNetEvents injects the OS network event handling.
|
||||
func WithNetEvents(events *netevents.Manager) ConnectClientOption {
|
||||
return func(c *ConnectClient) { c.netEvents = events }
|
||||
// WithNetworkState injects the OS network availability state that gates every
|
||||
// reconnection loop; without it gating is disabled.
|
||||
func WithNetworkState(netState *netstate.State) ConnectClientOption {
|
||||
return func(c *ConnectClient) { c.netState = netState }
|
||||
}
|
||||
|
||||
// WithSweeper injects the network change sweeper.
|
||||
func WithSweeper(sweeper *netsweep.Sweeper) ConnectClientOption {
|
||||
return func(c *ConnectClient) { c.sweeper = sweeper }
|
||||
}
|
||||
|
||||
func NewConnectClient(
|
||||
@@ -293,7 +305,7 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan
|
||||
}
|
||||
|
||||
// suspend connection attempts while the OS reports no usable network
|
||||
if waited, err := c.netEvents.Wait(c.ctx); err != nil {
|
||||
if waited, err := c.netState.Wait(c.ctx); err != nil {
|
||||
return nil
|
||||
} else if waited {
|
||||
backOff.Reset()
|
||||
@@ -311,7 +323,7 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan
|
||||
|
||||
log.Debugf("connecting to the Management service %s", c.config.ManagementURL.Host)
|
||||
mgmClient, err := mgm.NewClient(engineCtx, c.config.ManagementURL.Host, myPrivateKey, mgmTlsEnabled,
|
||||
mgm.WithNetEvents(c.netEvents))
|
||||
mgm.WithNetworkState(c.netState), mgm.WithSweeper(c.sweeper))
|
||||
if err != nil {
|
||||
// On daemon shutdown / Down() the parent context is cancelled
|
||||
// and the dial fails with "context canceled". Wrapping that
|
||||
@@ -386,7 +398,7 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan
|
||||
}()
|
||||
|
||||
// with the global Netbird config in hand connect (just a connection, no stream yet) Signal
|
||||
signalClient, err := connectToSignal(engineCtx, loginResp.GetNetbirdConfig(), myPrivateKey, c.netEvents)
|
||||
signalClient, err := connectToSignal(engineCtx, loginResp.GetNetbirdConfig(), myPrivateKey, c.netState, c.sweeper)
|
||||
if err != nil {
|
||||
log.Error(err)
|
||||
return wrapErr(err)
|
||||
@@ -423,7 +435,7 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan
|
||||
}
|
||||
|
||||
relayManager := relayClient.NewManager(engineCtx, relayURLs, myPrivateKey.PublicKey().String(), engineConfig.MTU,
|
||||
relayClient.WithNetEvents(c.netEvents))
|
||||
relayClient.WithNetworkState(c.netState), relayClient.WithSweeper(c.sweeper))
|
||||
c.statusRecorder.SetRelayMgr(relayManager)
|
||||
if len(relayURLs) > 0 {
|
||||
if token != nil {
|
||||
@@ -451,7 +463,7 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan
|
||||
UpdateManager: c.updateManager,
|
||||
ClientMetrics: c.clientMetrics,
|
||||
MetricsCtx: c.ctx,
|
||||
NetState: c.netEvents,
|
||||
NetState: c.netState,
|
||||
}, mobileDependency)
|
||||
engine.SetSyncResponsePersistence(c.persistSyncResponse)
|
||||
c.engine = engine
|
||||
@@ -711,7 +723,7 @@ func selectMTU(localMTU uint16, peerMTU int32) uint16 {
|
||||
}
|
||||
|
||||
// connectToSignal creates Signal Service client and established a connection
|
||||
func connectToSignal(ctx context.Context, wtConfig *mgmProto.NetbirdConfig, ourPrivateKey wgtypes.Key, netEvents *netevents.Manager) (*signal.GrpcClient, error) {
|
||||
func connectToSignal(ctx context.Context, wtConfig *mgmProto.NetbirdConfig, ourPrivateKey wgtypes.Key, netState *netstate.State, sweeper *netsweep.Sweeper) (*signal.GrpcClient, error) {
|
||||
var sigTLSEnabled bool
|
||||
if wtConfig.Signal.Protocol == mgmProto.HostConfig_HTTPS {
|
||||
sigTLSEnabled = true
|
||||
@@ -720,7 +732,7 @@ func connectToSignal(ctx context.Context, wtConfig *mgmProto.NetbirdConfig, ourP
|
||||
}
|
||||
|
||||
signalClient, err := signal.NewClient(ctx, wtConfig.Signal.Uri, ourPrivateKey, sigTLSEnabled,
|
||||
signal.WithNetEvents(netEvents))
|
||||
signal.WithNetworkState(netState), signal.WithSweeper(sweeper))
|
||||
if err != nil {
|
||||
log.Errorf("error while connecting to the Signal Exchange Service %s: %s", wtConfig.Signal.Uri, err)
|
||||
return nil, gstatus.Errorf(codes.FailedPrecondition, "failed connecting to Signal Service : %s", err)
|
||||
|
||||
@@ -59,7 +59,7 @@ import (
|
||||
"github.com/netbirdio/netbird/client/internal/syncstore"
|
||||
"github.com/netbirdio/netbird/client/internal/updater"
|
||||
"github.com/netbirdio/netbird/client/jobexec"
|
||||
"github.com/netbirdio/netbird/client/netevents"
|
||||
"github.com/netbirdio/netbird/client/netstate"
|
||||
cProto "github.com/netbirdio/netbird/client/proto"
|
||||
"github.com/netbirdio/netbird/client/system"
|
||||
nbdns "github.com/netbirdio/netbird/dns"
|
||||
@@ -184,7 +184,7 @@ type EngineServices struct {
|
||||
MetricsCtx context.Context
|
||||
// NetState gates the reconnection loops on OS-reported network
|
||||
// availability; nil disables gating.
|
||||
NetState *netevents.Manager
|
||||
NetState *netstate.State
|
||||
}
|
||||
|
||||
// Engine is a mechanism responsible for reacting on Signal and Management stream events and managing connections to the remote peers.
|
||||
@@ -210,7 +210,7 @@ type Engine struct {
|
||||
|
||||
// netState gates the peer reconnection guards on OS-reported network
|
||||
// availability; nil disables gating.
|
||||
netState *netevents.Manager
|
||||
netState *netstate.State
|
||||
|
||||
// STUNs is a list of STUN servers used by ICE
|
||||
STUNs []*stun.URI
|
||||
|
||||
@@ -26,7 +26,7 @@ import (
|
||||
"github.com/netbirdio/netbird/client/internal/portforward"
|
||||
"github.com/netbirdio/netbird/client/internal/rosenpass"
|
||||
"github.com/netbirdio/netbird/client/internal/stdnet"
|
||||
"github.com/netbirdio/netbird/client/netevents"
|
||||
"github.com/netbirdio/netbird/client/netstate"
|
||||
"github.com/netbirdio/netbird/route"
|
||||
relayClient "github.com/netbirdio/netbird/shared/relay/client"
|
||||
)
|
||||
@@ -97,7 +97,7 @@ type ConnConfig struct {
|
||||
|
||||
// NetworkState gates the reconnection guard on OS-reported network
|
||||
// availability; nil disables gating.
|
||||
NetworkState *netevents.Manager
|
||||
NetworkState *netstate.State
|
||||
}
|
||||
|
||||
type Conn struct {
|
||||
|
||||
@@ -6,6 +6,8 @@ import (
|
||||
|
||||
"github.com/cenkalti/backoff/v4"
|
||||
log "github.com/sirupsen/logrus"
|
||||
|
||||
"github.com/netbirdio/netbird/client/netstate"
|
||||
)
|
||||
|
||||
// ConnStatus represents the connection state as seen by the guard.
|
||||
@@ -22,12 +24,6 @@ const (
|
||||
|
||||
type connStatusFunc func() ConnStatus
|
||||
|
||||
// NetworkWatcher is the availability view the guard gates reconnects on.
|
||||
type NetworkWatcher interface {
|
||||
IsOnline() bool
|
||||
Changed() <-chan struct{}
|
||||
}
|
||||
|
||||
// Guard is responsible for the reconnection logic.
|
||||
// It will trigger to send an offer to the peer then has connection issues.
|
||||
// Watch these events:
|
||||
@@ -43,14 +39,14 @@ type Guard struct {
|
||||
srWatcher *SRWatcher
|
||||
// netState gates reconnect attempts on OS-reported network availability;
|
||||
// nil disables gating.
|
||||
netState NetworkWatcher
|
||||
netState *netstate.State
|
||||
relayedConnDisconnected chan struct{}
|
||||
iCEConnDisconnected chan struct{}
|
||||
}
|
||||
|
||||
// NewGuard creates a reconnection guard for a peer connection. A nil netState
|
||||
// disables network availability gating.
|
||||
func NewGuard(log *log.Entry, isConnectedFn connStatusFunc, timeout time.Duration, srWatcher *SRWatcher, netState NetworkWatcher) *Guard {
|
||||
func NewGuard(log *log.Entry, isConnectedFn connStatusFunc, timeout time.Duration, srWatcher *SRWatcher, netState *netstate.State) *Guard {
|
||||
return &Guard{
|
||||
log: log,
|
||||
isConnectedOnAllWay: isConnectedFn,
|
||||
@@ -108,17 +104,14 @@ func (g *Guard) reconnectLoopWithRetry(ctx context.Context, callback func()) {
|
||||
iceState := &iceRetryState{log: g.log}
|
||||
defer iceState.reset()
|
||||
|
||||
var netChanged <-chan struct{}
|
||||
if g.netState != nil {
|
||||
netChanged = g.netState.Changed()
|
||||
}
|
||||
netChanged := g.netState.Changed()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-tickerChannel:
|
||||
// skip attempts while the OS reports no usable network; the
|
||||
// netChanged case below resumes the loop once it returns
|
||||
if g.netState != nil && !g.netState.IsOnline() {
|
||||
if !g.netState.IsOnline() {
|
||||
continue
|
||||
}
|
||||
switch g.isConnectedOnAllWay() {
|
||||
|
||||
@@ -9,7 +9,7 @@ import (
|
||||
log "github.com/sirupsen/logrus"
|
||||
|
||||
"github.com/netbirdio/netbird/client/internal/peer/ice"
|
||||
"github.com/netbirdio/netbird/client/netevents/netstate"
|
||||
"github.com/netbirdio/netbird/client/netstate"
|
||||
)
|
||||
|
||||
// newTestGuardWithNetState builds a guard with a realistic MaxInterval: the
|
||||
|
||||
@@ -307,6 +307,16 @@ func startUIInSession(uiPath string, sessionID uint32) error {
|
||||
}
|
||||
}()
|
||||
|
||||
var env *uint16
|
||||
if err := windows.CreateEnvironmentBlock(&env, primaryToken, false); err != nil {
|
||||
return fmt.Errorf("create environment block: %w", err)
|
||||
}
|
||||
defer func() {
|
||||
if err := windows.DestroyEnvironmentBlock(env); err != nil {
|
||||
log.Warnf("failed to destroy environment block: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
// Prepare startup info
|
||||
var si windows.StartupInfo
|
||||
si.Cb = uint32(unsafe.Sizeof(si))
|
||||
@@ -329,7 +339,7 @@ func startUIInSession(uiPath string, sessionID uint32) error {
|
||||
nil,
|
||||
false,
|
||||
creationFlags,
|
||||
nil,
|
||||
env,
|
||||
nil,
|
||||
&si,
|
||||
&pi,
|
||||
|
||||
@@ -22,7 +22,8 @@ import (
|
||||
"github.com/netbirdio/netbird/client/internal/listener"
|
||||
"github.com/netbirdio/netbird/client/internal/peer"
|
||||
"github.com/netbirdio/netbird/client/internal/profilemanager"
|
||||
"github.com/netbirdio/netbird/client/netevents"
|
||||
"github.com/netbirdio/netbird/client/netstate"
|
||||
"github.com/netbirdio/netbird/client/netsweep"
|
||||
"github.com/netbirdio/netbird/client/system"
|
||||
"github.com/netbirdio/netbird/formatter"
|
||||
"github.com/netbirdio/netbird/route"
|
||||
@@ -83,10 +84,12 @@ type Client struct {
|
||||
onHostDnsFn func([]string)
|
||||
dnsManager dns.IosDnsManager
|
||||
loginComplete bool
|
||||
// netMgr outlives engine restarts: it mirrors the OS connectivity, not
|
||||
// the engine lifecycle. Run injects its state and sweeper into each new
|
||||
// ConnectClient.
|
||||
netMgr *netevents.Manager
|
||||
// netState outlives engine restarts: it mirrors the OS connectivity, not
|
||||
// the engine lifecycle. Run injects it into each new ConnectClient, which
|
||||
// distributes it to every reconnection loop.
|
||||
netState *netstate.State
|
||||
// sweeper also outlives engine restarts; NotifyNetworkChange sweeps it.
|
||||
sweeper *netsweep.Sweeper
|
||||
// preloadedConfig holds config loaded from JSON (used on tvOS where file writes are blocked)
|
||||
preloadedConfig *profilemanager.Config
|
||||
|
||||
@@ -97,7 +100,6 @@ type Client struct {
|
||||
|
||||
// NewClient instantiate a new Client
|
||||
func NewClient(cfgFile, stateFile, cacheDir, logFilePath, deviceName string, osVersion string, osName string, networkChangeListener NetworkChangeListener, dnsManager DnsManager) *Client {
|
||||
recorder := peer.NewRecorder("")
|
||||
return &Client{
|
||||
cfgFile: cfgFile,
|
||||
stateFile: stateFile,
|
||||
@@ -106,11 +108,12 @@ func NewClient(cfgFile, stateFile, cacheDir, logFilePath, deviceName string, osV
|
||||
deviceName: deviceName,
|
||||
osName: osName,
|
||||
osVersion: osVersion,
|
||||
recorder: recorder,
|
||||
recorder: peer.NewRecorder(""),
|
||||
ctxCancelLock: &sync.Mutex{},
|
||||
networkChangeListener: networkChangeListener,
|
||||
dnsManager: dnsManager,
|
||||
netMgr: netevents.NewManager(recorder),
|
||||
netState: netstate.New(),
|
||||
sweeper: netsweep.New(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -187,7 +190,7 @@ func (c *Client) Run(fd int32, interfaceName string, envList *EnvList) error {
|
||||
cfg.WgIface = interfaceName
|
||||
|
||||
connectClient := internal.NewConnectClient(ctx, cfg, c.recorder,
|
||||
internal.WithNetEvents(c.netMgr))
|
||||
internal.WithNetworkState(c.netState), internal.WithSweeper(c.sweeper))
|
||||
c.setState(cfg, connectClient)
|
||||
// Persist the latest sync response so DebugBundle can include the network
|
||||
// map. On iOS this is backed by disk to keep it out of the constrained
|
||||
@@ -200,11 +203,10 @@ func (c *Client) Run(fd int32, interfaceName string, envList *EnvList) error {
|
||||
// (e.g. from NWPathMonitor). While unavailable, the internal reconnect loops
|
||||
// suspend their attempts and the connection listener reports NoNetwork
|
||||
// instead of Connecting; when availability returns, the loops resume
|
||||
// immediately with a fresh backoff. Losing the last network also sweeps the
|
||||
// registered connections, so the client does not keep reporting Connected
|
||||
// over stale sockets with no network at all.
|
||||
// immediately with a fresh backoff.
|
||||
func (c *Client) SetNetworkAvailable(available bool) {
|
||||
c.netMgr.SetNetworkAvailable(available)
|
||||
c.netState.Set(available)
|
||||
c.recorder.SetNetworkAvailable(available)
|
||||
}
|
||||
|
||||
// NotifyNetworkChange marks the management, signal and relay connections
|
||||
@@ -212,7 +214,8 @@ func (c *Client) SetNetworkAvailable(available bool) {
|
||||
// whatever has not redialed on the new network by then. The engine and the
|
||||
// TUN device stay untouched.
|
||||
func (c *Client) NotifyNetworkChange() {
|
||||
c.netMgr.NotifyNetworkChange()
|
||||
c.sweeper.MarkNetworkChange()
|
||||
log.Infof("network change: connections marked stale")
|
||||
}
|
||||
|
||||
// Stop the internal client and free the resources
|
||||
|
||||
@@ -1,158 +0,0 @@
|
||||
// Package netevents owns the OS network event handling shared by the mobile
|
||||
// bindings: availability changes park or wake the reconnection loops and drive
|
||||
// the NoNetwork listener state, and both losing the last network and switching
|
||||
// networks sweep the stale connections so their owners redial immediately.
|
||||
package netevents
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/cenkalti/backoff/v4"
|
||||
log "github.com/sirupsen/logrus"
|
||||
|
||||
"github.com/netbirdio/netbird/client/netevents/netstate"
|
||||
"github.com/netbirdio/netbird/client/netevents/sweep"
|
||||
)
|
||||
|
||||
// Recorder receives the availability changes for listener state reporting.
|
||||
type Recorder interface {
|
||||
SetNetworkAvailable(available bool)
|
||||
}
|
||||
|
||||
// Manager ties the network availability state, the connection sweeper and the
|
||||
// status recorder together; it outlives engine restarts. A nil *Manager is
|
||||
// the valid no-events value: the read methods report always-online and never
|
||||
// sweep.
|
||||
type Manager struct {
|
||||
netState *netstate.State
|
||||
sweeper *sweep.Sweeper
|
||||
recorder Recorder
|
||||
}
|
||||
|
||||
// NewManager creates a Manager reporting into recorder, starting online.
|
||||
func NewManager(recorder Recorder) *Manager {
|
||||
return &Manager{
|
||||
netState: netstate.New(),
|
||||
sweeper: sweep.New(),
|
||||
recorder: recorder,
|
||||
}
|
||||
}
|
||||
|
||||
// SetNetworkAvailable records OS-reported network availability. While
|
||||
// unavailable, the reconnection loops suspend their attempts and the
|
||||
// connection listener reports NoNetwork instead of Connecting; when
|
||||
// availability returns, the loops resume immediately with a fresh backoff.
|
||||
// Losing the last network also sweeps the registered connections: nothing can
|
||||
// redial while offline, so the stale sockets would otherwise stay silently
|
||||
// "connected" until their own timeouts and the client would keep reporting
|
||||
// Connected with no network at all.
|
||||
func (m *Manager) SetNetworkAvailable(available bool) {
|
||||
if !available && m.netState.IsOnline() {
|
||||
m.sweeper.MarkNetworkChange()
|
||||
}
|
||||
m.netState.Set(available)
|
||||
m.recorder.SetNetworkAvailable(available)
|
||||
}
|
||||
|
||||
// NotifyNetworkChange marks the management, signal and relay connections
|
||||
// stale after the OS switched networks and schedules a sweep that cuts
|
||||
// whatever has not redialed on the new network by then. The engine and the
|
||||
// TUN device stay untouched.
|
||||
func (m *Manager) NotifyNetworkChange() {
|
||||
m.sweeper.MarkNetworkChange()
|
||||
log.Infof("network change: connections marked stale")
|
||||
}
|
||||
|
||||
// IsOnline reports whether the OS reports at least one usable network.
|
||||
func (m *Manager) IsOnline() bool {
|
||||
if m == nil {
|
||||
return true
|
||||
}
|
||||
return m.netState.IsOnline()
|
||||
}
|
||||
|
||||
// Changed returns a channel closed on the next availability transition.
|
||||
func (m *Manager) Changed() <-chan struct{} {
|
||||
if m == nil {
|
||||
return nil
|
||||
}
|
||||
return m.netState.Changed()
|
||||
}
|
||||
|
||||
// Wait blocks while the network is offline; see netstate.State.Wait.
|
||||
func (m *Manager) Wait(ctx context.Context) (bool, error) {
|
||||
if m == nil {
|
||||
return false, nil
|
||||
}
|
||||
return m.netState.Wait(ctx)
|
||||
}
|
||||
|
||||
// WaitSettled waits until an online verdict holds for a full settleWindow, or
|
||||
// while offline until the budget runs out. Returns false when ctx is
|
||||
// cancelled. The settle window exists because a disconnect often precedes the
|
||||
// OS offline flag by a few milliseconds, so a fresh online verdict cannot be
|
||||
// trusted immediately. A nil Manager has no events to watch: it degrades to a
|
||||
// fixed budget-long sleep.
|
||||
func (m *Manager) WaitSettled(ctx context.Context, budget, settleWindow time.Duration) bool {
|
||||
if m == nil {
|
||||
select {
|
||||
case <-time.After(budget):
|
||||
return true
|
||||
case <-ctx.Done():
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
budgetTimer := time.NewTimer(budget)
|
||||
defer budgetTimer.Stop()
|
||||
|
||||
settle := time.NewTimer(settleWindow)
|
||||
defer settle.Stop()
|
||||
|
||||
for {
|
||||
// Channel first, flag second: a flip in between still fires the channel.
|
||||
changedCh := m.netState.Changed()
|
||||
if m.netState.IsOnline() {
|
||||
select {
|
||||
case <-settle.C:
|
||||
return true
|
||||
case <-changedCh:
|
||||
case <-ctx.Done():
|
||||
return false
|
||||
}
|
||||
} else {
|
||||
select {
|
||||
case <-budgetTimer.C:
|
||||
return true
|
||||
case <-changedCh:
|
||||
case <-ctx.Done():
|
||||
return false
|
||||
}
|
||||
}
|
||||
if !settle.Stop() {
|
||||
select {
|
||||
case <-settle.C:
|
||||
default:
|
||||
}
|
||||
}
|
||||
settle.Reset(settleWindow)
|
||||
}
|
||||
}
|
||||
|
||||
// StartDial registers an in-flight dial with the sweeper; see sweep.Sweeper.StartDial.
|
||||
func (m *Manager) StartDial(ctx context.Context) *sweep.Dial {
|
||||
if m == nil {
|
||||
return (*sweep.Sweeper)(nil).StartDial(ctx)
|
||||
}
|
||||
return m.sweeper.StartDial(ctx)
|
||||
}
|
||||
|
||||
// QuickRetryBackoff wraps bo for a quick retry after a network change; see
|
||||
// sweep.Sweeper.QuickRetryBackoff.
|
||||
func (m *Manager) QuickRetryBackoff(ctx context.Context, bo backoff.BackOff) backoff.BackOff {
|
||||
if m == nil {
|
||||
return bo
|
||||
}
|
||||
return m.sweeper.QuickRetryBackoff(ctx, bo, m.netState)
|
||||
}
|
||||
@@ -1,34 +0,0 @@
|
||||
package netevents
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
type recorderStub struct{}
|
||||
|
||||
func (recorderStub) SetNetworkAvailable(bool) {}
|
||||
|
||||
func TestWaitSettledAfterOutage(t *testing.T) {
|
||||
const budget = 1500 * time.Millisecond
|
||||
const settleWindow = 200 * time.Millisecond
|
||||
const outage = 2 * settleWindow
|
||||
|
||||
m := NewManager(recorderStub{})
|
||||
m.SetNetworkAvailable(false)
|
||||
|
||||
start := time.Now()
|
||||
go func() {
|
||||
time.Sleep(outage)
|
||||
m.SetNetworkAvailable(true)
|
||||
}()
|
||||
|
||||
ok := m.WaitSettled(context.Background(), budget, settleWindow)
|
||||
elapsed := time.Since(start)
|
||||
|
||||
assert.True(t, ok, "recovered network must let the caller proceed")
|
||||
assert.GreaterOrEqual(t, elapsed, outage+settleWindow, "an online verdict must hold a full settle window before it is trusted")
|
||||
}
|
||||
@@ -1,10 +1,10 @@
|
||||
// Package sweep cuts network-bound activity when the OS switches networks:
|
||||
// Package netsweep cuts network-bound activity when the OS switches networks:
|
||||
// a sweep closes the registered connections and aborts the in-flight dials, so
|
||||
// their owners redial immediately instead of waiting for the old sockets to
|
||||
// time out.
|
||||
//
|
||||
// A nil *Sweeper disables everything: all methods are nil-safe no-ops.
|
||||
package sweep
|
||||
package netsweep
|
||||
|
||||
import (
|
||||
"context"
|
||||
@@ -16,7 +16,7 @@ import (
|
||||
"github.com/cenkalti/backoff/v4"
|
||||
log "github.com/sirupsen/logrus"
|
||||
|
||||
"github.com/netbirdio/netbird/client/netevents/netstate"
|
||||
"github.com/netbirdio/netbird/client/netstate"
|
||||
)
|
||||
|
||||
// DefaultSweepDelay absorbs network flapping while the OS settles on a
|
||||
@@ -34,7 +34,7 @@ type Config struct {
|
||||
// ErrSwept reports that a dial finished after a network change swept its
|
||||
// registration. The connection is already closed; the caller must treat it
|
||||
// as a failed dial and redial on the new network.
|
||||
var ErrSwept = errors.New("sweep: connection swept by network change")
|
||||
var ErrSwept = errors.New("netsweep: connection swept by network change")
|
||||
|
||||
// sweepID identifies one registration in a sweeper. Connections and dials
|
||||
// draw from the same counter, so an id is unique across both registries.
|
||||
@@ -1,4 +1,4 @@
|
||||
package sweep
|
||||
package netsweep
|
||||
|
||||
import (
|
||||
"context"
|
||||
@@ -1,11 +1,11 @@
|
||||
package sweep
|
||||
package netsweep
|
||||
|
||||
import (
|
||||
"time"
|
||||
|
||||
"github.com/cenkalti/backoff/v4"
|
||||
|
||||
"github.com/netbirdio/netbird/client/netevents/netstate"
|
||||
"github.com/netbirdio/netbird/client/netstate"
|
||||
)
|
||||
|
||||
const quickRetryDelay = 200 * time.Millisecond
|
||||
@@ -1,4 +1,4 @@
|
||||
package sweep
|
||||
package netsweep
|
||||
|
||||
import (
|
||||
"context"
|
||||
@@ -21,7 +21,8 @@ import (
|
||||
"google.golang.org/grpc/connectivity"
|
||||
|
||||
nbgrpc "github.com/netbirdio/netbird/client/grpc"
|
||||
"github.com/netbirdio/netbird/client/netevents"
|
||||
"github.com/netbirdio/netbird/client/netstate"
|
||||
"github.com/netbirdio/netbird/client/netsweep"
|
||||
"github.com/netbirdio/netbird/client/system"
|
||||
"github.com/netbirdio/netbird/encryption"
|
||||
"github.com/netbirdio/netbird/shared/management/domain"
|
||||
@@ -63,9 +64,12 @@ type GrpcClient struct {
|
||||
connStateCallbackLock sync.RWMutex
|
||||
serverURL string
|
||||
|
||||
// netEvents gates the stream retry loop on OS-reported network
|
||||
// availability and sweeps the transport on network change.
|
||||
netEvents *netevents.Manager
|
||||
// netState gates the stream retry loop on OS-reported network
|
||||
// availability; nil (the default) disables gating.
|
||||
netState *netstate.State
|
||||
|
||||
// sweeper cuts the transport connections on network change; nil disables it.
|
||||
sweeper *netsweep.Sweeper
|
||||
|
||||
// syncStreamErr holds the last Sync stream error, or nil while the stream
|
||||
// is established and healthy. GetServerKey succeeds even when the peer
|
||||
@@ -119,9 +123,15 @@ func MaxRecvMsgSize() int {
|
||||
// Option configures optional GrpcClient behavior.
|
||||
type Option func(*GrpcClient)
|
||||
|
||||
// WithNetEvents injects the OS network event handling.
|
||||
func WithNetEvents(events *netevents.Manager) Option {
|
||||
return func(c *GrpcClient) { c.netEvents = events }
|
||||
// WithNetworkState injects the OS network availability state that gates the
|
||||
// stream retry loop; without it gating is disabled.
|
||||
func WithNetworkState(netState *netstate.State) Option {
|
||||
return func(c *GrpcClient) { c.netState = netState }
|
||||
}
|
||||
|
||||
// WithSweeper injects the network change sweeper.
|
||||
func WithSweeper(sweeper *netsweep.Sweeper) Option {
|
||||
return func(c *GrpcClient) { c.sweeper = sweeper }
|
||||
}
|
||||
|
||||
// NewClient creates a new client to Management service
|
||||
@@ -142,7 +152,9 @@ func NewClient(ctx context.Context, addr string, ourPrivateKey wgtypes.Key, tlsE
|
||||
extraOpts = append(extraOpts, grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(maxSize)))
|
||||
log.Infof("management gRPC max receive message size set to %d bytes", maxSize)
|
||||
}
|
||||
extraOpts = append(extraOpts, nbgrpc.WithSweeper(c.netEvents))
|
||||
if c.sweeper != nil {
|
||||
extraOpts = append(extraOpts, nbgrpc.WithSweeper(c.sweeper))
|
||||
}
|
||||
|
||||
var conn *grpc.ClientConn
|
||||
operation := func() error {
|
||||
@@ -223,19 +235,16 @@ func (c *GrpcClient) withMgmtStream(
|
||||
ctx context.Context,
|
||||
handler func(ctx context.Context, serverPubKey wgtypes.Key, backOff backoff.BackOff) error,
|
||||
) error {
|
||||
backOff := c.netEvents.QuickRetryBackoff(ctx, 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
|
||||
// stop the loop without reporting a failure.
|
||||
if waited, err := c.netEvents.Wait(ctx); err != nil {
|
||||
if waited, err := c.netState.Wait(ctx); err != nil {
|
||||
log.Debugf("management connection context has been canceled while offline, this usually indicates shutdown")
|
||||
return nil //nolint:nilerr // a cancelled context means shutdown, not a retryable failure
|
||||
} else if waited {
|
||||
backOff.Reset()
|
||||
// dials attempted while offline grew the channel's internal backoff;
|
||||
// reset it too, or the reconnect waits out that timer first
|
||||
c.conn.ResetConnectBackoff()
|
||||
}
|
||||
|
||||
connState := c.conn.GetState()
|
||||
@@ -264,7 +273,7 @@ func (c *GrpcClient) withMgmtStream(
|
||||
return handler(ctx, *serverPubKey, backOff)
|
||||
}
|
||||
|
||||
err := nbgrpc.Retry(ctx, operation, backOff, c.netEvents)
|
||||
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)
|
||||
}
|
||||
|
||||
@@ -14,7 +14,7 @@ import (
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
|
||||
"github.com/netbirdio/netbird/client/netevents/sweep"
|
||||
"github.com/netbirdio/netbird/client/netsweep"
|
||||
auth "github.com/netbirdio/netbird/shared/relay/auth/hmac"
|
||||
"github.com/netbirdio/netbird/shared/relay/client/dialer"
|
||||
netErr "github.com/netbirdio/netbird/shared/relay/client/dialer/net"
|
||||
@@ -151,14 +151,6 @@ type transportConn interface {
|
||||
Protocol() string
|
||||
}
|
||||
|
||||
// NetEvents is the OS network event view the relay consumes: availability
|
||||
// gating for the reconnect guard and dial registration for the network change
|
||||
// sweep.
|
||||
type NetEvents interface {
|
||||
NetworkWatcher
|
||||
StartDial(ctx context.Context) *sweep.Dial
|
||||
}
|
||||
|
||||
// Client is a client for the relay server. It is responsible for establishing a connection to the relay server and
|
||||
// managing connections to other peers. All exported functions are safe to call concurrently. After close the connection,
|
||||
// the client can be reused by calling Connect again. When the client is closed, all connections are closed too.
|
||||
@@ -194,10 +186,9 @@ type Client struct {
|
||||
// the manager.
|
||||
transportFallback *transportFallback
|
||||
|
||||
// netEvents registers the relay dial for the network change sweep; the
|
||||
// read loop reports the disconnect and the guard reconnects. Shared via
|
||||
// the manager.
|
||||
netEvents NetEvents
|
||||
// sweeper cuts the relay connection on network change; the read loop
|
||||
// reports the disconnect and the guard reconnects. Shared via the manager.
|
||||
sweeper *netsweep.Sweeper
|
||||
// datagramFallbackTriggered guards a single fallback per connection so a
|
||||
// burst of oversized datagrams triggers one reconnect, not many.
|
||||
datagramFallbackTriggered atomic.Bool
|
||||
@@ -409,12 +400,7 @@ func (c *Client) Close() error {
|
||||
func (c *Client) connect(ctx context.Context) (*RelayAddr, error) {
|
||||
// A sweep cancels this context, so a dial started on the old network
|
||||
// aborts instead of waiting out its handshake timeout.
|
||||
var dial *sweep.Dial
|
||||
if c.netEvents != nil {
|
||||
dial = c.netEvents.StartDial(ctx)
|
||||
} else {
|
||||
dial = (*sweep.Sweeper)(nil).StartDial(ctx)
|
||||
}
|
||||
dial := c.sweeper.StartDial(ctx)
|
||||
defer dial.Release()
|
||||
ctx = dial.Ctx()
|
||||
|
||||
|
||||
@@ -7,6 +7,8 @@ import (
|
||||
|
||||
"github.com/cenkalti/backoff/v4"
|
||||
log "github.com/sirupsen/logrus"
|
||||
|
||||
"github.com/netbirdio/netbird/client/netstate"
|
||||
)
|
||||
|
||||
const (
|
||||
@@ -22,13 +24,6 @@ const (
|
||||
verdictSettleWindow = 200 * time.Millisecond
|
||||
)
|
||||
|
||||
// NetworkWatcher is the availability view the guard gates reconnects on.
|
||||
type NetworkWatcher interface {
|
||||
Wait(ctx context.Context) (bool, error)
|
||||
IsOnline() bool
|
||||
WaitSettled(ctx context.Context, budget, settleWindow time.Duration) bool
|
||||
}
|
||||
|
||||
// Guard manage the reconnection tries to the Relay server in case of disconnection event.
|
||||
type Guard struct {
|
||||
// OnNewRelayClient is a channel that is used to notify the relay manager about a new relay client instance.
|
||||
@@ -40,8 +35,9 @@ type Guard struct {
|
||||
// attempts.
|
||||
maxBackoffInterval time.Duration
|
||||
|
||||
// netState gates reconnect attempts on OS-reported network availability.
|
||||
netState NetworkWatcher
|
||||
// netState gates reconnect attempts on OS-reported network availability;
|
||||
// nil disables gating.
|
||||
netState *netstate.State
|
||||
|
||||
// lastErr is the error from the most recent failed reconnect attempt,
|
||||
// surfaced as the home relay status while disconnected.
|
||||
@@ -49,8 +45,9 @@ type Guard struct {
|
||||
}
|
||||
|
||||
// NewGuard creates a new guard for the relay client. A non-positive
|
||||
// maxBackoffInterval falls back to defaultMaxBackoffInterval.
|
||||
func NewGuard(sp *ServerPicker, maxBackoffInterval time.Duration, netState NetworkWatcher) *Guard {
|
||||
// maxBackoffInterval falls back to defaultMaxBackoffInterval. A nil netState
|
||||
// disables network availability gating.
|
||||
func NewGuard(sp *ServerPicker, maxBackoffInterval time.Duration, netState *netstate.State) *Guard {
|
||||
if maxBackoffInterval <= 0 {
|
||||
maxBackoffInterval = defaultMaxBackoffInterval
|
||||
}
|
||||
@@ -100,14 +97,12 @@ func (g *Guard) StartReconnectTrys(ctx context.Context, relayClient *Client) {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
// suspend reconnect attempts while the OS reports no usable network
|
||||
if g.netState != nil {
|
||||
if waited, err := g.netState.Wait(ctx); err != nil {
|
||||
return
|
||||
} else if waited {
|
||||
ticker.Stop()
|
||||
ticker = g.exponentTicker(ctx)
|
||||
continue
|
||||
}
|
||||
if waited, err := g.netState.Wait(ctx); err != nil {
|
||||
return
|
||||
} else if waited {
|
||||
ticker.Stop()
|
||||
ticker = g.exponentTicker(ctx)
|
||||
continue
|
||||
}
|
||||
if err := g.retry(ctx); err != nil {
|
||||
log.Errorf("failed to pick new Relay server: %s", err)
|
||||
@@ -134,18 +129,13 @@ func (g *Guard) tryToQuickReconnect(parentCtx context.Context, rc *Client) bool
|
||||
return false
|
||||
}
|
||||
|
||||
if g.netState != nil {
|
||||
if ok := g.netState.WaitSettled(parentCtx, quickReconnectBudget, verdictSettleWindow); !ok {
|
||||
return false
|
||||
}
|
||||
// Still offline after the budget: leave the retry to the ticker.
|
||||
if !g.netState.IsOnline() {
|
||||
return false
|
||||
}
|
||||
} else {
|
||||
if cancelled := waiteBeforeRetry(parentCtx); !cancelled {
|
||||
return false
|
||||
}
|
||||
if ok := g.waitForNetwork(parentCtx); !ok {
|
||||
return false
|
||||
}
|
||||
|
||||
// Still offline after the budget: leave the retry to the ticker.
|
||||
if !g.netState.IsOnline() {
|
||||
return false
|
||||
}
|
||||
|
||||
log.Infof("try to reconnect to Relay server: %s", rc.connectionURL)
|
||||
@@ -210,14 +200,47 @@ func (g *Guard) exponentTicker(ctx context.Context) *backoff.Ticker {
|
||||
return backoff.NewTicker(bo)
|
||||
}
|
||||
|
||||
func waiteBeforeRetry(ctx context.Context) bool {
|
||||
timer := time.NewTimer(quickReconnectBudget)
|
||||
defer timer.Stop()
|
||||
// waitForNetwork waits out the settle window while online, or waits for the
|
||||
// network to return while offline, within the budget. Returns false when ctx
|
||||
// is cancelled. Without an injected netState it degrades to a fixed
|
||||
// budget-long sleep, the pre-netstate behavior.
|
||||
func (g *Guard) waitForNetwork(ctx context.Context) bool {
|
||||
budget := time.NewTimer(quickReconnectBudget)
|
||||
defer budget.Stop()
|
||||
|
||||
select {
|
||||
case <-timer.C:
|
||||
return true
|
||||
case <-ctx.Done():
|
||||
return false
|
||||
settleWindow := verdictSettleWindow
|
||||
if g.netState == nil {
|
||||
settleWindow = quickReconnectBudget
|
||||
}
|
||||
settle := time.NewTimer(settleWindow)
|
||||
defer settle.Stop()
|
||||
|
||||
for {
|
||||
// Channel first, flag second: a flip in between still fires the channel.
|
||||
changedCh := g.netState.Changed()
|
||||
if g.netState.IsOnline() {
|
||||
select {
|
||||
case <-settle.C:
|
||||
return true
|
||||
case <-changedCh:
|
||||
case <-ctx.Done():
|
||||
return false
|
||||
}
|
||||
} else {
|
||||
select {
|
||||
case <-budget.C:
|
||||
return true
|
||||
case <-changedCh:
|
||||
case <-ctx.Done():
|
||||
return false
|
||||
}
|
||||
}
|
||||
if !settle.Stop() {
|
||||
select {
|
||||
case <-settle.C:
|
||||
default:
|
||||
}
|
||||
}
|
||||
settle.Reset(settleWindow)
|
||||
}
|
||||
}
|
||||
|
||||
30
shared/relay/client/guard_test.go
Normal file
30
shared/relay/client/guard_test.go
Normal file
@@ -0,0 +1,30 @@
|
||||
package client
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
|
||||
"github.com/netbirdio/netbird/client/netstate"
|
||||
)
|
||||
|
||||
func TestWaitForNetworkSettlesAfterOutage(t *testing.T) {
|
||||
ns := netstate.New()
|
||||
ns.Set(false)
|
||||
g := NewGuard(nil, 0, ns)
|
||||
|
||||
const outage = 2 * verdictSettleWindow
|
||||
start := time.Now()
|
||||
go func() {
|
||||
time.Sleep(outage)
|
||||
ns.Set(true)
|
||||
}()
|
||||
|
||||
ok := g.waitForNetwork(context.Background())
|
||||
elapsed := time.Since(start)
|
||||
|
||||
assert.True(t, ok, "recovered network must let the quick reconnect proceed")
|
||||
assert.GreaterOrEqual(t, elapsed, outage+verdictSettleWindow, "reconnect must wait a full settle window after the network returns")
|
||||
}
|
||||
@@ -12,6 +12,8 @@ import (
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
|
||||
"github.com/netbirdio/netbird/client/netstate"
|
||||
"github.com/netbirdio/netbird/client/netsweep"
|
||||
relayAuth "github.com/netbirdio/netbird/shared/relay/auth/hmac"
|
||||
)
|
||||
|
||||
@@ -65,9 +67,15 @@ func WithMaxBackoffInterval(d time.Duration) ManagerOption {
|
||||
return func(m *Manager) { m.maxBackoffInterval = d }
|
||||
}
|
||||
|
||||
// WithNetEvents injects the OS network event handling.
|
||||
func WithNetEvents(events NetEvents) ManagerOption {
|
||||
return func(m *Manager) { m.netEvents = events }
|
||||
// WithNetworkState injects the OS network availability state that gates the
|
||||
// reconnect guard; without it reconnect attempts are not gated.
|
||||
func WithNetworkState(netState *netstate.State) ManagerOption {
|
||||
return func(m *Manager) { m.netState = netState }
|
||||
}
|
||||
|
||||
// WithSweeper injects the network change sweeper.
|
||||
func WithSweeper(sweeper *netsweep.Sweeper) ManagerOption {
|
||||
return func(m *Manager) { m.sweeper = sweeper }
|
||||
}
|
||||
|
||||
// Manager is a manager for the relay client instances. It establishes one persistent connection to the given relay URL
|
||||
@@ -97,7 +105,8 @@ type Manager struct {
|
||||
|
||||
mtu uint16
|
||||
maxBackoffInterval time.Duration
|
||||
netEvents NetEvents
|
||||
netState *netstate.State
|
||||
sweeper *netsweep.Sweeper
|
||||
|
||||
cleanupInterval time.Duration
|
||||
keepUnusedServerTime time.Duration
|
||||
@@ -134,9 +143,9 @@ func NewManager(ctx context.Context, serverURLs []string, peerID string, mtu uin
|
||||
for _, opt := range opts {
|
||||
opt(m)
|
||||
}
|
||||
m.serverPicker.NetEvents = m.netEvents
|
||||
m.serverPicker.Sweeper = m.sweeper
|
||||
m.serverPicker.ServerURLs.Store(serverURLs)
|
||||
m.reconnectGuard = NewGuard(m.serverPicker, m.maxBackoffInterval, m.netEvents)
|
||||
m.reconnectGuard = NewGuard(m.serverPicker, m.maxBackoffInterval, m.netState)
|
||||
return m
|
||||
}
|
||||
|
||||
@@ -361,7 +370,7 @@ func (m *Manager) openConnVia(ctx context.Context, serverAddress, peerKey string
|
||||
|
||||
relayClient := NewClientWithServerIP(serverAddress, serverIP, m.tokenStore, m.peerID, m.mtu)
|
||||
relayClient.SetTransportFallback(m.transportFallback)
|
||||
relayClient.netEvents = m.netEvents
|
||||
relayClient.sweeper = m.sweeper
|
||||
err := relayClient.Connect(m.ctx)
|
||||
if err != nil {
|
||||
rt.Lock()
|
||||
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
|
||||
"github.com/netbirdio/netbird/client/netsweep"
|
||||
auth "github.com/netbirdio/netbird/shared/relay/auth/hmac"
|
||||
)
|
||||
|
||||
@@ -30,7 +31,7 @@ type ServerPicker struct {
|
||||
MTU uint16
|
||||
ConnectionTimeout time.Duration
|
||||
TransportFallback *transportFallback
|
||||
NetEvents NetEvents
|
||||
Sweeper *netsweep.Sweeper
|
||||
}
|
||||
|
||||
func (sp *ServerPicker) PickServer(parentCtx context.Context) (*Client, error) {
|
||||
@@ -74,7 +75,7 @@ func (sp *ServerPicker) startConnection(ctx context.Context, resultChan chan con
|
||||
log.Infof("try to connecting to relay server: %s", url)
|
||||
relayClient := NewClient(url, sp.TokenStore, sp.PeerID, sp.MTU)
|
||||
relayClient.SetTransportFallback(sp.TransportFallback)
|
||||
relayClient.netEvents = sp.NetEvents
|
||||
relayClient.sweeper = sp.Sweeper
|
||||
err := relayClient.Connect(ctx)
|
||||
resultChan <- connResult{
|
||||
RelayClient: relayClient,
|
||||
|
||||
@@ -19,7 +19,8 @@ import (
|
||||
"google.golang.org/grpc/status"
|
||||
|
||||
nbgrpc "github.com/netbirdio/netbird/client/grpc"
|
||||
"github.com/netbirdio/netbird/client/netevents"
|
||||
"github.com/netbirdio/netbird/client/netstate"
|
||||
"github.com/netbirdio/netbird/client/netsweep"
|
||||
"github.com/netbirdio/netbird/encryption"
|
||||
"github.com/netbirdio/netbird/shared/management/client"
|
||||
"github.com/netbirdio/netbird/shared/signal/proto"
|
||||
@@ -66,9 +67,12 @@ type GrpcClient struct {
|
||||
connStateCallback ConnStateNotifier
|
||||
connStateCallbackLock sync.RWMutex
|
||||
|
||||
// netEvents gates the Receive retry loop on OS-reported network
|
||||
// availability and sweeps the transport on network change.
|
||||
netEvents *netevents.Manager
|
||||
// netState gates the Receive retry loop on OS-reported network
|
||||
// availability; nil (the default) disables gating.
|
||||
netState *netstate.State
|
||||
|
||||
// sweeper cuts the transport connections on network change; nil disables it.
|
||||
sweeper *netsweep.Sweeper
|
||||
|
||||
onReconnectedListenerFn func()
|
||||
|
||||
@@ -96,9 +100,15 @@ type GrpcClient struct {
|
||||
// Option configures optional GrpcClient behavior.
|
||||
type Option func(*GrpcClient)
|
||||
|
||||
// WithNetEvents injects the OS network event handling.
|
||||
func WithNetEvents(events *netevents.Manager) Option {
|
||||
return func(c *GrpcClient) { c.netEvents = events }
|
||||
// WithNetworkState injects the OS network availability state that gates the
|
||||
// Receive retry loop; without it gating is disabled.
|
||||
func WithNetworkState(netState *netstate.State) Option {
|
||||
return func(c *GrpcClient) { c.netState = netState }
|
||||
}
|
||||
|
||||
// WithSweeper injects the network change sweeper.
|
||||
func WithSweeper(sweeper *netsweep.Sweeper) Option {
|
||||
return func(c *GrpcClient) { c.sweeper = sweeper }
|
||||
}
|
||||
|
||||
// NewClient creates a new Signal client
|
||||
@@ -116,7 +126,9 @@ func NewClient(ctx context.Context, addr string, key wgtypes.Key, tlsEnabled boo
|
||||
}
|
||||
|
||||
var extraOpts []grpc.DialOption
|
||||
extraOpts = append(extraOpts, nbgrpc.WithSweeper(c.netEvents))
|
||||
if c.sweeper != nil {
|
||||
extraOpts = append(extraOpts, nbgrpc.WithSweeper(c.sweeper))
|
||||
}
|
||||
|
||||
var conn *grpc.ClientConn
|
||||
operation := func() error {
|
||||
@@ -186,20 +198,17 @@ 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 {
|
||||
|
||||
backOff := c.netEvents.QuickRetryBackoff(ctx, 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
|
||||
// stop the loop without reporting a failure.
|
||||
if waited, err := c.netEvents.Wait(ctx); err != nil {
|
||||
if waited, err := c.netState.Wait(ctx); err != nil {
|
||||
log.Debugf("signal connection context has been canceled while offline, this usually indicates shutdown")
|
||||
return nil
|
||||
} else if waited {
|
||||
backOff.Reset()
|
||||
// dials attempted while offline grew the channel's internal backoff;
|
||||
// reset it too, or the reconnect waits out that timer first
|
||||
c.signalConn.ResetConnectBackoff()
|
||||
}
|
||||
|
||||
c.notifyStreamDisconnected()
|
||||
@@ -272,7 +281,7 @@ func (c *GrpcClient) Receive(ctx context.Context, msgHandler func(msg *proto.Mes
|
||||
return nil
|
||||
}
|
||||
|
||||
err := nbgrpc.Retry(ctx, operation, backOff, c.netEvents)
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user