From f62949605e2b758bd5874e46432a932fdf5201cc Mon Sep 17 00:00:00 2001 From: Viktor Liu <17948409+lixmal@users.noreply.github.com> Date: Tue, 13 Oct 2026 23:13:25 +0900 Subject: [PATCH] [client] Keep HA routing peer selection latency-aware and hold switches that reset connections (#7689) --- client/internal/peer/status.go | 173 +++- client/internal/peer/status_test.go | 178 +++++ client/internal/peer/worker_ice.go | 161 +++- client/internal/peer/worker_ice_test.go | 168 ++++ client/internal/routemanager/client/client.go | 755 ++++++++++++++---- .../routemanager/client/client_bench_test.go | 8 +- .../routemanager/client/client_test.go | 652 +++++++++++++++ client/internal/routemanager/client/env.go | 77 ++ .../client/relayed_switch_test.go | 358 +++++++++ 9 files changed, 2334 insertions(+), 196 deletions(-) create mode 100644 client/internal/peer/worker_ice_test.go create mode 100644 client/internal/routemanager/client/env.go create mode 100644 client/internal/routemanager/client/relayed_switch_test.go diff --git a/client/internal/peer/status.go b/client/internal/peer/status.go index 6c44178e1..b695e3b22 100644 --- a/client/internal/peer/status.go +++ b/client/internal/peer/status.go @@ -28,6 +28,11 @@ import ( const eventQueueSize = 10 +// routerLatencyNotifyThreshold is the latency or noise change that notifies the route +// watchers subscribed to a peer. Latency is resampled continuously, so without +// a threshold every sample would wake every watcher of every routing peer. +const routerLatencyNotifyThreshold = 5 * time.Millisecond + type ResolvedDomainInfo struct { Prefixes []netip.Prefix ParentDomain domain.Domain @@ -41,11 +46,21 @@ type EventListener interface { OnEvent(event *proto.SystemEvent) } +// LatencySample is a peer's smoothed round trip time together with the +// uncertainty of that estimate, which is what the jitter of the path leaves +// behind after smoothing. Comparing two peers is only meaningful when their +// difference is larger than this. +type LatencySample struct { + Latency time.Duration + Noise time.Duration +} + // RouterState status for router peers. This contains relevant fields for route manager type RouterState struct { - Status ConnStatus - Relayed bool - Latency time.Duration + Status ConnStatus + Relayed bool + Latency time.Duration + LatencyNoise time.Duration } // State contains the latest state of a peer @@ -67,9 +82,11 @@ type State struct { BytesTx int64 BytesRx int64 Latency time.Duration - RosenpassEnabled bool - SSHHostKey []byte - routes map[string]struct{} + // LatencyNoise is the uncertainty of Latency on a jittery path + LatencyNoise time.Duration + RosenpassEnabled bool + SSHHostKey []byte + routes map[string]struct{} } // AddRoute add a single route to routes map @@ -103,6 +120,13 @@ func (s *State) GetRoutes() map[string]struct{} { return maps.Clone(s.routes) } +// clearLatency drops the latency estimate, which is only measured over a direct +// connection and says nothing about the path once that connection is gone. +func (s *State) clearLatency() { + s.Latency = 0 + s.LatencyNoise = 0 +} + // LocalPeerState contains the latest state of the local peer type LocalPeerState struct { IP string @@ -163,11 +187,25 @@ type FullStatus struct { Events []*proto.SystemEvent } +// routerSnapshot is a router-state map stamped with a sequence number taken +// under the status lock. Snapshots are built under the lock but delivered +// outside it, so two dispatchers can race on the way to a subscriber; the +// stamp lets delivery drop the older snapshot instead of letting it overwrite +// a newer one. +type routerSnapshot struct { + seq uint64 + states map[string]RouterState +} + type StatusChangeSubscription struct { peerID string id string eventsChan chan map[string]RouterState ctx context.Context + + // sendMu serializes deliveries and guards lastSeq + sendMu sync.Mutex + lastSeq uint64 } func newStatusChangeSubscription(ctx context.Context, peerID string) *StatusChangeSubscription { @@ -184,17 +222,54 @@ func (s *StatusChangeSubscription) Events() chan map[string]RouterState { return s.eventsChan } +// deliver forwards a snapshot to the subscriber, dropping it when a newer one +// has already been delivered. A blocking deliver waits for buffer space, since +// peer state transitions must not be lost. A non-blocking one gives up when the +// subscriber is backlogged or another delivery is in flight; it is meant for +// latency refreshes, which are superseded by the next sample anyway and are +// also re-read from the recorder on the route watchers' periodic re-evaluation. +// Only a snapshot that reached the subscriber supersedes older ones. +func (s *StatusChangeSubscription) deliver(snapshot *routerSnapshot, block bool) { + if block { + s.sendMu.Lock() + } else if !s.sendMu.TryLock() { + return + } + defer s.sendMu.Unlock() + + if snapshot.seq <= s.lastSeq { + return + } + + if block { + select { + case s.eventsChan <- snapshot.states: + s.lastSeq = snapshot.seq + case <-s.ctx.Done(): + } + return + } + + select { + case s.eventsChan <- snapshot.states: + s.lastSeq = snapshot.seq + default: + } +} + // Status holds a state of peers, signal, management connections and relays. // mux is an RWMutex so hot read paths (notably PeerStateByIP, called for // every private-service request) don't contend against each other. // Pure read methods take RLock; anything that mutates state takes Lock. type Status struct { - mux sync.RWMutex - muxRelays sync.RWMutex - peers map[string]State - ipToKey map[string]string - activeRoutePeers map[route.HAUniqueID]string - changeNotify map[string]map[string]*StatusChangeSubscription // map[peerID]map[subscriptionID]*StatusChangeSubscription + mux sync.RWMutex + muxRelays sync.RWMutex + peers map[string]State + ipToKey map[string]string + activeRoutePeers map[route.HAUniqueID]string + changeNotify map[string]map[string]*StatusChangeSubscription // map[peerID]map[subscriptionID]*StatusChangeSubscription + // routerSeq stamps router-state snapshots, see routerSnapshot + routerSeq uint64 signalState bool signalError error managementState bool @@ -386,6 +461,7 @@ func (d *Status) UpdatePeerState(receivedState State) error { peerState.RemoteIceCandidateEndpoint = receivedState.RemoteIceCandidateEndpoint peerState.RelayServerAddress = receivedState.RelayServerAddress peerState.RosenpassEnabled = receivedState.RosenpassEnabled + peerState.clearLatency() } d.peers[receivedState.PubKey] = peerState @@ -402,7 +478,7 @@ func (d *Status) UpdatePeerState(receivedState State) error { d.notifier.peerListChanged(numPeers) } if notifyRouter { - d.dispatchRouterPeers(receivedState.PubKey, routerSnapshot) + d.dispatchRouterPeers(receivedState.PubKey, routerSnapshot, true) } d.notifyStateChange() return nil @@ -522,7 +598,7 @@ func (d *Status) UpdatePeerICEState(receivedState State) error { d.notifier.peerListChanged(numPeers) } if notifyRouter { - d.dispatchRouterPeers(receivedState.PubKey, routerSnapshot) + d.dispatchRouterPeers(receivedState.PubKey, routerSnapshot, true) } d.notifyStateChange() return nil @@ -559,7 +635,7 @@ func (d *Status) UpdatePeerRelayedState(receivedState State) error { d.notifier.peerListChanged(numPeers) } if notifyRouter { - d.dispatchRouterPeers(receivedState.PubKey, routerSnapshot) + d.dispatchRouterPeers(receivedState.PubKey, routerSnapshot, true) } d.notifyStateChange() return nil @@ -595,7 +671,7 @@ func (d *Status) UpdatePeerRelayedStateToDisconnected(receivedState State) error d.notifier.peerListChanged(numPeers) } if notifyRouter { - d.dispatchRouterPeers(receivedState.PubKey, routerSnapshot) + d.dispatchRouterPeers(receivedState.PubKey, routerSnapshot, true) } d.notifyStateChange() return nil @@ -620,6 +696,7 @@ func (d *Status) UpdatePeerICEStateToDisconnected(receivedState State) error { peerState.RemoteIceCandidateType = receivedState.RemoteIceCandidateType peerState.LocalIceCandidateEndpoint = receivedState.LocalIceCandidateEndpoint peerState.RemoteIceCandidateEndpoint = receivedState.RemoteIceCandidateEndpoint + peerState.clearLatency() d.peers[receivedState.PubKey] = peerState @@ -634,7 +711,7 @@ func (d *Status) UpdatePeerICEStateToDisconnected(receivedState State) error { d.notifier.peerListChanged(numPeers) } if notifyRouter { - d.dispatchRouterPeers(receivedState.PubKey, routerSnapshot) + d.dispatchRouterPeers(receivedState.PubKey, routerSnapshot, true) } d.notifyStateChange() return nil @@ -714,7 +791,7 @@ func (d *Status) FinishPeerListModifications() { // snapshot per-peer router state to deliver after the lock is released type routerDispatch struct { peerID string - snapshot map[string]RouterState + snapshot *routerSnapshot } dispatches := make([]routerDispatch, 0, len(d.peers)) for key := range d.peers { @@ -728,7 +805,7 @@ func (d *Status) FinishPeerListModifications() { d.notifier.peerListChanged(numPeers) for _, rd := range dispatches { - d.dispatchRouterPeers(rd.peerID, rd.snapshot) + d.dispatchRouterPeers(rd.peerID, rd.snapshot, true) } d.notifyStateChange() } @@ -1057,22 +1134,42 @@ func (d *Status) GetManagementState() ManagementState { } } -func (d *Status) UpdateLatency(pubKey string, latency time.Duration) error { - if latency <= 0 { +func (d *Status) UpdateLatency(pubKey string, sample LatencySample) error { + if sample.Latency <= 0 { return nil } d.mux.Lock() - defer d.mux.Unlock() peerState, ok := d.peers[pubKey] if !ok { + d.mux.Unlock() return errors.New("peer doesn't exist") } - peerState.Latency = latency + + previous, previousNoise := peerState.Latency, peerState.LatencyNoise + peerState.Latency = sample.Latency + peerState.LatencyNoise = sample.Noise d.peers[pubKey] = peerState + + // the noise sets the switch margin, so it counts on its own + notifyRouter := absDuration(sample.Latency-previous) >= routerLatencyNotifyThreshold || + absDuration(sample.Noise-previousNoise) >= routerLatencyNotifyThreshold + routerSnapshot := d.snapshotRouterPeersLocked(pubKey, notifyRouter) + + d.mux.Unlock() + + d.dispatchRouterPeers(pubKey, routerSnapshot, false) + return nil } +func absDuration(d time.Duration) time.Duration { + if d < 0 { + return -d + } + return d +} + // IsLoginRequired determines if a peer's login has expired. func (d *Status) IsLoginRequired() bool { d.mux.RLock() @@ -1229,7 +1326,7 @@ func (d *Status) RemoveConnectionListener() { // Caller MUST hold d.mux. Returns nil when there are no subscribers for peerID // or when notify is false. The snapshot is consumed later by dispatchRouterPeers // outside the lock so the channel send cannot stall any d.mux holder. -func (d *Status) snapshotRouterPeersLocked(peerID string, notify bool) map[string]RouterState { +func (d *Status) snapshotRouterPeersLocked(peerID string, notify bool) *routerSnapshot { if !notify { return nil } @@ -1244,22 +1341,29 @@ func (d *Status) snapshotRouterPeersLocked(peerID string, notify bool) map[strin continue } routerPeers[pid] = RouterState{ - Status: s.ConnStatus, - Relayed: s.Relayed, - Latency: s.Latency, + Status: s.ConnStatus, + Relayed: s.Relayed, + Latency: s.Latency, + LatencyNoise: s.LatencyNoise, } } - return routerPeers + + d.routerSeq++ + + return &routerSnapshot{seq: d.routerSeq, states: routerPeers} } // dispatchRouterPeers delivers a previously snapshotted router-state map to // the peer's subscribers. Caller MUST NOT hold d.mux. The method takes a // fresh, short read of d.changeNotify under the lock to grab subscriber // channels, then sends outside the lock so a slow consumer cannot block other -// d.mux holders. The send itself stays blocking (only short-circuited by the -// subscriber's context) so peer state transitions are not silently dropped. -func (d *Status) dispatchRouterPeers(peerID string, routerPeers map[string]RouterState) { - if routerPeers == nil { +// d.mux holders. With block set, the send waits for buffer space (only +// short-circuited by the subscriber's context) so peer state transitions are +// not silently dropped; without it, a backlogged subscriber loses the snapshot +// instead of stalling the caller. Out-of-order deliveries are dropped either +// way, see StatusChangeSubscription.deliver. +func (d *Status) dispatchRouterPeers(peerID string, snapshot *routerSnapshot, block bool) { + if snapshot == nil { return } @@ -1274,10 +1378,7 @@ func (d *Status) dispatchRouterPeers(peerID string, routerPeers map[string]Route d.mux.Unlock() for _, sub := range subs { - select { - case sub.eventsChan <- routerPeers: - case <-sub.ctx.Done(): - } + sub.deliver(snapshot, block) } } diff --git a/client/internal/peer/status_test.go b/client/internal/peer/status_test.go index b3f01b217..2d910ccf5 100644 --- a/client/internal/peer/status_test.go +++ b/client/internal/peer/status_test.go @@ -44,6 +44,184 @@ func TestGetPeer(t *testing.T) { assert.Error(t, err, "should return error when peer doesn't exist") } +func TestUpdateLatencyNotifiesRouteWatchers(t *testing.T) { + key := "abc" + status := NewRecorder("https://mgm") + require.NoError(t, status.AddPeer(key, "abc.netbird", "100.108.254.1", "")) + + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + sub := status.SubscribeToPeerStateChanges(ctx, key) + + require.NoError(t, status.UpdateLatency(key, LatencySample{Latency: 40 * time.Millisecond})) + select { + case states := <-sub.Events(): + assert.Equal(t, 40*time.Millisecond, states[key].Latency, "the notified state should carry the new latency") + default: + t.Fatal("expected a router state notification for the first latency sample") + } + + // below routerLatencyNotifyThreshold: recorded, but not worth waking the watchers + require.NoError(t, status.UpdateLatency(key, LatencySample{Latency: 42 * time.Millisecond})) + select { + case <-sub.Events(): + t.Fatal("a latency change below the threshold should not notify") + default: + } + + peerState, err := status.GetPeer(key) + require.NoError(t, err) + assert.Equal(t, 42*time.Millisecond, peerState.Latency, "the latency should be recorded even without a notification") + + require.NoError(t, status.UpdateLatency(key, LatencySample{Latency: 100 * time.Millisecond})) + select { + case states := <-sub.Events(): + assert.Equal(t, 100*time.Millisecond, states[key].Latency, "a significant latency change should notify") + default: + t.Fatal("expected a router state notification for a significant latency change") + } + + // the noise feeds the switch margin, so a large change in it matters on its + // own even when the latency stays put + require.NoError(t, status.UpdateLatency(key, LatencySample{Latency: 100 * time.Millisecond, Noise: 20 * time.Millisecond})) + select { + case states := <-sub.Events(): + assert.Equal(t, 20*time.Millisecond, states[key].LatencyNoise, "a significant noise change should notify") + default: + t.Fatal("expected a router state notification for a significant noise change") + } + + require.NoError(t, status.UpdateLatency(key, LatencySample{Latency: 100 * time.Millisecond, Noise: 21 * time.Millisecond})) + select { + case <-sub.Events(): + t.Fatal("a noise change below the threshold should not notify") + default: + } +} + +// TestSubscriptionDeliverOrdering covers the sequence stamping: snapshots are +// built under the status lock but delivered outside it, so a dispatcher that +// is descheduled between the two can deliver an outdated snapshot after a +// newer one. The older snapshot must be dropped, not delivered. +func TestSubscriptionDeliverOrdering(t *testing.T) { + sub := newStatusChangeSubscription(context.Background(), "peer") + + newer := &routerSnapshot{seq: 2, states: map[string]RouterState{"peer": {Status: StatusConnecting}}} + older := &routerSnapshot{seq: 1, states: map[string]RouterState{"peer": {Status: StatusConnected}}} + + sub.deliver(newer, true) + sub.deliver(older, true) + + select { + case states := <-sub.Events(): + assert.Equal(t, StatusConnecting, states["peer"].Status, "the newer snapshot should be delivered") + default: + t.Fatal("expected the newer snapshot to be delivered") + } + + select { + case <-sub.Events(): + t.Fatal("the outdated snapshot must be dropped, not delivered after the newer one") + default: + } +} + +// TestSubscriptionDeliverNonBlocking covers the latency dispatch path: a +// backlogged subscriber must not block the sender, and the dropped snapshot is +// recovered from the recorder by the route watchers' periodic re-evaluation. +func TestSubscriptionDeliverNonBlocking(t *testing.T) { + sub := newStatusChangeSubscription(context.Background(), "peer") + + // fill the subscription buffer with no consumer draining it + seq := uint64(1) + for { + before := len(sub.eventsChan) + sub.deliver(&routerSnapshot{seq: seq, states: map[string]RouterState{}}, false) + seq++ + if len(sub.eventsChan) == before { + break + } + } + + done := make(chan struct{}) + go func() { + sub.deliver(&routerSnapshot{seq: seq, states: map[string]RouterState{}}, false) + close(done) + }() + + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("a non-blocking deliver must return with the buffer full") + } +} + +// A snapshot the subscriber never received must not count as delivered: a +// state transition dispatched just before it, but delivered just after, would +// otherwise be dropped as outdated and never reach the subscriber. +func TestSubscriptionDroppedDeliveryDoesNotSupersede(t *testing.T) { + sub := newStatusChangeSubscription(context.Background(), "peer") + + for seq := uint64(1); seq <= uint64(cap(sub.eventsChan)); seq++ { + sub.deliver(&routerSnapshot{seq: seq, states: map[string]RouterState{}}, false) + } + require.Len(t, sub.eventsChan, cap(sub.eventsChan), "the buffer must be full") + + // a latency refresh stamped after the transition wins the race to the + // subscriber and is dropped on the full buffer + transition := &routerSnapshot{seq: 100, states: map[string]RouterState{"peer": {Status: StatusConnecting}}} + latency := &routerSnapshot{seq: 101, states: map[string]RouterState{"peer": {Status: StatusConnected}}} + sub.deliver(latency, false) + + for range cap(sub.eventsChan) { + <-sub.eventsChan + } + sub.deliver(transition, true) + + select { + case states := <-sub.Events(): + assert.Equal(t, StatusConnecting, states["peer"].Status, "the transition must be delivered") + default: + t.Fatal("a transition must not be dropped because a later snapshot was dropped") + } +} + +// Latency is only measured over a direct connection. Once it is gone the last +// sample describes a path the traffic no longer takes, and must not be compared +// against other routing peers as if it were current. +func TestLatencyClearedWithDirectConnection(t *testing.T) { + const key = "abc" + + newStatus := func(t *testing.T) *Status { + t.Helper() + status := NewRecorder("https://mgm") + require.NoError(t, status.AddPeer(key, "peer-a.netbird.local", "10.10.10.10", "")) + require.NoError(t, status.UpdatePeerICEState(State{PubKey: key, ConnStatus: StatusConnected})) + require.NoError(t, status.UpdateLatency(key, LatencySample{Latency: 20 * time.Millisecond, Noise: time.Millisecond})) + return status + } + + assertCleared := func(t *testing.T, status *Status) { + t.Helper() + state, err := status.GetPeer(key) + require.NoError(t, err) + assert.Zero(t, state.Latency, "the latency of the lost direct path must be cleared") + assert.Zero(t, state.LatencyNoise, "the noise of the lost direct path must be cleared") + } + + t.Run("ice disconnected, relayed", func(t *testing.T) { + status := newStatus(t) + require.NoError(t, status.UpdatePeerICEStateToDisconnected(State{PubKey: key, ConnStatus: StatusConnected, Relayed: true})) + assertCleared(t, status) + }) + + t.Run("connection closed", func(t *testing.T) { + status := newStatus(t) + require.NoError(t, status.UpdatePeerState(State{PubKey: key, ConnStatus: StatusIdle})) + assertCleared(t, status) + }) +} + func TestUpdatePeerState(t *testing.T) { key := "abc" ip := "10.10.10.10" diff --git a/client/internal/peer/worker_ice.go b/client/internal/peer/worker_ice.go index 5979e9bdc..257164bbd 100644 --- a/client/internal/peer/worker_ice.go +++ b/client/internal/peer/worker_ice.go @@ -3,6 +3,7 @@ package peer import ( "context" "fmt" + "math" "net" "strconv" "sync" @@ -20,6 +21,26 @@ import ( "github.com/netbirdio/netbird/route" ) +const ( + // latencySampleInterval is how often the round trip time of the selected + // candidate pair is read. The ICE agent refreshes it on every consent + // binding request, so sampling is cheap and needs no extra traffic. + latencySampleInterval = 4 * time.Second + + // latencyEWMAAlpha is the weight of a new sample in the exponentially + // weighted moving average. At latencySampleInterval this yields a time + // constant of roughly 27 seconds: a sustained change is reflected within + // tens of seconds, while the residual movement on a jittery path stays + // below the switch margin that route selection applies, so noise alone + // cannot move a route. Raising it widens that residual movement. + latencyEWMAAlpha = 0.15 +) + +// latencySmoothedNoiseFactor converts the spread of the raw samples into the +// spread of the smoothed value: for an exponentially weighted moving average +// with weight a, the variance of the output is a/(2-a) of the input variance. +var latencySmoothedNoiseFactor = math.Sqrt(latencyEWMAAlpha / (2 - latencyEWMAAlpha)) + type ICEConnInfo struct { RemoteConn net.Conn RosenpassPubKey []byte @@ -137,7 +158,7 @@ func (w *WorkerICE) OnNewOffer(remoteOfferAnswer *OfferAnswer) { w.log.Debugf("recreate ICE agent: %s / %s", w.sessionID, *remoteOfferAnswer.SessionID) } dialerCtx, dialerCancel := context.WithCancel(w.ctx) - agent, err := w.reCreateAgent(dialerCancel, preferredCandidateTypes) + agent, sampleNow, err := w.reCreateAgent(dialerCancel, preferredCandidateTypes) if err != nil { w.log.Errorf("failed to recreate ICE Agent: %s", err) return @@ -153,7 +174,7 @@ func (w *WorkerICE) OnNewOffer(remoteOfferAnswer *OfferAnswer) { // Capture the cancel func at spawn time: connect reads it from the argument // instead of the field, which a newer OnNewOffer may already have replaced. - go w.connect(dialerCtx, dialerCancel, agent, remoteOfferAnswer) + go w.connect(dialerCtx, dialerCancel, agent, sampleNow, remoteOfferAnswer) } // OnRemoteCandidate Handles ICE connection Candidate provided by the remote peer. @@ -217,29 +238,33 @@ func (w *WorkerICE) Close() { w.abandonNegotiation() } -func (w *WorkerICE) reCreateAgent(dialerCancel context.CancelFunc, candidates []ice.CandidateType) (*icemaker.ThreadSafeAgent, error) { +// reCreateAgent creates a new ICE agent with its callbacks registered. The +// returned channel wakes that agent's latency sampler when its selected +// candidate pair changes. +func (w *WorkerICE) reCreateAgent(dialerCancel context.CancelFunc, candidates []ice.CandidateType) (*icemaker.ThreadSafeAgent, chan struct{}, error) { w.portForwardAttempted = false agent, err := icemaker.NewAgent(w.ctx, w.iFaceDiscover, w.config.ICEConfig, candidates, w.localUfrag, w.localPwd) if err != nil { - return nil, fmt.Errorf("create agent: %w", err) + return nil, nil, fmt.Errorf("create agent: %w", err) } if err := agent.OnCandidate(w.onICECandidate); err != nil { - return nil, err + return nil, nil, err } if err := agent.OnConnectionStateChange(w.onConnectionStateChange(agent, dialerCancel)); err != nil { - return nil, err + return nil, nil, err } - if err := agent.OnSelectedCandidatePairChange(func(c1, c2 ice.Candidate) { - w.onICESelectedCandidatePair(agent, c1, c2) - }); err != nil { - return nil, err + // one channel per agent, so the sampler of a replaced agent cannot take a + // wakeup meant for the new one + sampleNow := make(chan struct{}, 1) + if err := agent.OnSelectedCandidatePairChange(w.onICESelectedCandidatePair(sampleNow)); err != nil { + return nil, nil, err } - return agent, nil + return agent, sampleNow, nil } func (w *WorkerICE) SessionID() ICESessionID { @@ -252,7 +277,7 @@ func (w *WorkerICE) SessionID() ICESessionID { // will block until connection succeeded // but it won't release if ICE Agent went into Disconnected or Failed state, // so we have to cancel it with the provided context once agent detected a broken connection -func (w *WorkerICE) connect(ctx context.Context, dialerCancel context.CancelFunc, agent *icemaker.ThreadSafeAgent, remoteOfferAnswer *OfferAnswer) { +func (w *WorkerICE) connect(ctx context.Context, dialerCancel context.CancelFunc, agent *icemaker.ThreadSafeAgent, sampleNow <-chan struct{}, remoteOfferAnswer *OfferAnswer) { w.log.Debugf("gather candidates") if err := agent.GatherCandidates(); err != nil { w.log.Warnf("failed to gather candidates: %s", err) @@ -343,6 +368,9 @@ func (w *WorkerICE) connect(ctx context.Context, dialerCancel context.CancelFunc w.lastSuccess = time.Now() w.muxAgent.Unlock() + // ctx is the agent's dialer context, so sampling stops when the agent is closed + go w.sampleLatency(ctx, agent, sampleNow) + // todo: the potential problem is a race between the onConnectionStateChange // and the delivery below: after this unlock, a newer offer can replace // w.agent before onICEConnectionIsReady runs, delivering this (now stale) @@ -532,21 +560,63 @@ func (w *WorkerICE) createForwardedCandidate(srflxCandidate ice.Candidate, mappi return candidate, nil } -func (w *WorkerICE) onICESelectedCandidatePair(agent *icemaker.ThreadSafeAgent, c1, c2 ice.Candidate) { - w.log.Debugf("selected candidate pair [local <-> remote] -> [%s <-> %s], peer %s", c1.String(), c2.String(), - w.config.Key) +func (w *WorkerICE) onICESelectedCandidatePair(sampleNow chan<- struct{}) func(c1, c2 ice.Candidate) { + return func(c1, c2 ice.Candidate) { + w.log.Debugf("selected candidate pair [local <-> remote] -> [%s <-> %s], peer %s", c1.String(), c2.String(), + w.config.Key) + // refresh the recorded latency right away: it still describes the previous + // path, and route selection may run before the next sampling tick + select { + case sampleNow <- struct{}{}: + default: + } + } +} + +// sampleLatency records the smoothed round trip time of the agent's selected +// candidate pair until the agent is closed. Route selection compares the +// recorded latencies of high availability peers, so it has to keep following +// the path instead of freezing at the value observed when the pair was chosen. +func (w *WorkerICE) sampleLatency(ctx context.Context, agent *icemaker.ThreadSafeAgent, sampleNow <-chan struct{}) { + var sampler latencySampler + + ticker := time.NewTicker(latencySampleInterval) + defer ticker.Stop() + + for { + w.updateLatency(ctx, agent, &sampler) + + select { + case <-ctx.Done(): + return + case <-ticker.C: + case <-sampleNow: + } + } +} + +func (w *WorkerICE) updateLatency(ctx context.Context, agent *icemaker.ThreadSafeAgent, sampler *latencySampler) { pairStat, ok := agent.GetSelectedCandidatePairStats() if !ok { - w.log.Warnf("failed to get selected candidate pair stats") + w.log.Debugf("failed to get selected candidate pair stats") return } - duration := time.Duration(pairStat.CurrentRoundTripTime * float64(time.Second)) - if err := w.statusRecorder.UpdateLatency(w.config.Key, duration); err != nil { - w.log.Debugf("failed to update latency for peer: %s", err) + sample := sampler.observe(pairStat) + if sample.Latency <= 0 { return } + + // the agent may have been closed while sampling, and its connection's + // latency cleared: a late sample must not bring the stale value back + if ctx.Err() != nil { + return + } + + if err := w.statusRecorder.UpdateLatency(w.config.Key, sample); err != nil { + w.log.Debugf("failed to update latency for peer: %s", err) + } } func (w *WorkerICE) logSuccessfulPaths(agent *icemaker.ThreadSafeAgent) { @@ -612,6 +682,59 @@ func (w *WorkerICE) agentDial(ctx context.Context, agent *icemaker.ThreadSafeAge } } +// latencySampler smooths the round trip time samples of a single ICE agent and +// tracks how much they scatter, so consumers can tell a real difference between +// two paths from the noise of a jittery one. +type latencySampler struct { + smoothed time.Duration + deviation time.Duration + localID string + remoteID string +} + +// observe folds a candidate pair stat into the moving average and returns the +// smoothed latency with the uncertainty of that estimate. A different candidate +// pair starts over: the path changed, so the previous samples describe a route +// that no longer carries the traffic. +func (s *latencySampler) observe(stat ice.CandidatePairStats) LatencySample { + rtt := time.Duration(stat.CurrentRoundTripTime * float64(time.Second)) + if rtt <= 0 { + return LatencySample{} + } + + if s.smoothed <= 0 || s.localID != stat.LocalCandidateID || s.remoteID != stat.RemoteCandidateID { + s.localID = stat.LocalCandidateID + s.remoteID = stat.RemoteCandidateID + s.smoothed = rtt + // A fresh path is described by a single raw sample, which is exactly + // when the estimate deserves the least trust: a lucky outlier reported + // with no uncertainty could move a route on its own. Seed the deviation + // at half the observed value so the reported noise stays conservative + // until real samples accumulate, and keep a larger deviation carried + // over from the previous path, since the link's jitter usually outlives + // a candidate pair change. + if seed := rtt / 2; s.deviation < seed { + s.deviation = seed + } + return s.sample() + } + + // mean absolute deviation of the samples around the average, the same + // estimator TCP keeps next to its smoothed round trip time + delta := absDuration(rtt - s.smoothed) + s.deviation = time.Duration(latencyEWMAAlpha*float64(delta) + (1-latencyEWMAAlpha)*float64(s.deviation)) + s.smoothed = time.Duration(latencyEWMAAlpha*float64(rtt) + (1-latencyEWMAAlpha)*float64(s.smoothed)) + + return s.sample() +} + +func (s *latencySampler) sample() LatencySample { + return LatencySample{ + Latency: s.smoothed, + Noise: time.Duration(latencySmoothedNoiseFactor * float64(s.deviation)), + } +} + func shouldAddExtraCandidate(candidate ice.Candidate) bool { if candidate.Type() != ice.CandidateTypeServerReflexive { return false diff --git a/client/internal/peer/worker_ice_test.go b/client/internal/peer/worker_ice_test.go new file mode 100644 index 000000000..3089841c2 --- /dev/null +++ b/client/internal/peer/worker_ice_test.go @@ -0,0 +1,168 @@ +package peer + +import ( + "testing" + "time" + + "github.com/pion/ice/v4" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func pairStat(local, remote string, rtt time.Duration) ice.CandidatePairStats { + return ice.CandidatePairStats{ + LocalCandidateID: local, + RemoteCandidateID: remote, + CurrentRoundTripTime: rtt.Seconds(), + } +} + +func TestLatencySamplerObserve(t *testing.T) { + t.Run("first sample is taken as is", func(t *testing.T) { + var s latencySampler + + got := s.observe(pairStat("local", "remote", 40*time.Millisecond)) + assert.Equal(t, 40*time.Millisecond, got.Latency, "the first sample has nothing to average with") + }) + + t.Run("zero round trip time is not recorded", func(t *testing.T) { + var s latencySampler + + require.Zero(t, s.observe(pairStat("local", "remote", 0)).Latency, "a pair without an RTT sample yields no latency") + + // the first real sample still seeds the average + assert.Equal(t, 20*time.Millisecond, s.observe(pairStat("local", "remote", 20*time.Millisecond)).Latency) + }) + + t.Run("single spike is damped", func(t *testing.T) { + var s latencySampler + + s.observe(pairStat("local", "remote", 20*time.Millisecond)) + spiked := s.observe(pairStat("local", "remote", 220*time.Millisecond)).Latency + + // a single sample only contributes its weight of the 200ms delta + assert.Equal(t, 50*time.Millisecond, spiked, "a single spike must not be reported in full") + + // and it decays again once the path is back to normal + for range 10 { + s.observe(pairStat("local", "remote", 20*time.Millisecond)) + } + assert.Less(t, s.observe(pairStat("local", "remote", 20*time.Millisecond)).Latency, 30*time.Millisecond, + "the average should return towards the steady state") + }) + + t.Run("sustained change is followed", func(t *testing.T) { + var s latencySampler + + s.observe(pairStat("local", "remote", 20*time.Millisecond)) + + // the route selection switch margin, see routemanager/client + const switchMargin = 20 * time.Millisecond + + // at a 4s sample interval, 3 samples are about 12 seconds + var latency time.Duration + for range 3 { + latency = s.observe(pairStat("local", "remote", 80*time.Millisecond)).Latency + } + assert.Greater(t, latency-20*time.Millisecond, switchMargin, + "a sustained 60ms degradation must clear the switch margin within seconds") + + // and 10 more samples, about 40 seconds in total, get most of the way there + for range 10 { + latency = s.observe(pairStat("local", "remote", 80*time.Millisecond)).Latency + } + assert.Greater(t, latency, 70*time.Millisecond, "a sustained degradation must be tracked within tens of seconds") + }) + + // Smoothing narrows the scatter of a jittery path but leaves a residue, and + // that residue is what route selection has to compare its margins against. + // The reported noise has to describe it, so it has to grow with the jitter + // and cover the movement of the smoothed value. + t.Run("reported noise reflects the jitter of the path", func(t *testing.T) { + // 100ms +/- 40ms, the same shape the netem jitter test uses + jittery := []int{ + 104, 138, 61, 96, 141, 72, 118, 65, 133, 88, + 59, 127, 92, 136, 68, 111, 143, 74, 99, 121, + } + steady := []int{ + 100, 101, 99, 100, 102, 99, 100, 101, 100, 99, 101, 100, 99, 100, 101, 100, 99, 101, 100, 100, + 101, 99, 100, 102, 99, 100, 101, 100, 99, 101, 100, 99, 100, 101, 100, 99, 101, 100, 100, 101, + } + + spreadAndNoise := func(raw []int) (spread, noise time.Duration) { + var s latencySampler + var minSmoothed, maxSmoothed time.Duration + + for i, ms := range raw { + got := s.observe(pairStat("local", "remote", time.Duration(ms)*time.Millisecond)) + noise = got.Noise + + // skip the warmup, the first sample is taken as is + if i < 5 { + continue + } + if minSmoothed == 0 || got.Latency < minSmoothed { + minSmoothed = got.Latency + } + if got.Latency > maxSmoothed { + maxSmoothed = got.Latency + } + } + + return maxSmoothed - minSmoothed, noise + } + + jitterySpread, jitteryNoise := spreadAndNoise(jittery) + steadySpread, steadyNoise := spreadAndNoise(steady) + + assert.Greater(t, jitteryNoise, steadyNoise, "a jittery path must report more noise than a steady one") + assert.Less(t, steadyNoise, time.Millisecond, "a steady path must report almost no noise, so the fixed margins decide") + + // the noise is a deviation, so a few of them have to cover the movement + // of the smoothed value, which is what the switch margin relies on + assert.Greater(t, 4*jitteryNoise, jitterySpread, + "reported noise %v must account for the smoothed spread %v", jitteryNoise, jitterySpread) + assert.Greater(t, 4*steadyNoise+time.Millisecond, steadySpread, + "reported noise %v must account for the smoothed spread %v", steadyNoise, steadySpread) + }) + + t.Run("new candidate pair discards the previous average", func(t *testing.T) { + var s latencySampler + + s.observe(pairStat("local", "remote", 200*time.Millisecond)) + got := s.observe(pairStat("local2", "remote2", 20*time.Millisecond)).Latency + + assert.Equal(t, 20*time.Millisecond, got, "samples of the previous path must not be carried over") + }) + + // A single raw sample is when the estimate deserves the least trust: with + // zero reported noise, a lucky outlier could clear the switch margins and + // move a route on its own. + t.Run("a fresh path reports conservative noise", func(t *testing.T) { + var s latencySampler + + first := s.observe(pairStat("local", "remote", 100*time.Millisecond)) + assert.Greater(t, first.Noise, 10*time.Millisecond, "a single-sample estimate must not be reported as noise-free") + + // steady samples shrink the uncertainty again + var got LatencySample + for range 40 { + got = s.observe(pairStat("local", "remote", 100*time.Millisecond)) + } + assert.Less(t, got.Noise, 2*time.Millisecond, "steady samples should earn the estimate its trust back") + }) + + t.Run("a pair change keeps the larger deviation of a jittery link", func(t *testing.T) { + var s latencySampler + + // build up a deviation well above the reseed floor of the new path + for _, ms := range []int{100, 180, 60, 170, 50, 190, 70, 160} { + s.observe(pairStat("local", "remote", time.Duration(ms)*time.Millisecond)) + } + before := s.deviation + require.Greater(t, before, 25*time.Millisecond, "the jittery path should have accumulated deviation") + + s.observe(pairStat("local2", "remote2", 50*time.Millisecond)) + assert.GreaterOrEqual(t, s.deviation, before, "a pair change must not shrink the known jitter of the link") + }) +} diff --git a/client/internal/routemanager/client/client.go b/client/internal/routemanager/client/client.go index 973cf1ab8..b882c4d4a 100644 --- a/client/internal/routemanager/client/client.go +++ b/client/internal/routemanager/client/client.go @@ -3,8 +3,10 @@ package client import ( "context" "fmt" + "math" "net/netip" "reflect" + "sync" "time" log "github.com/sirupsen/logrus" @@ -26,6 +28,48 @@ const ( handlerTypeStatic ) +const ( + // latencySwitchMinGain is the absolute latency improvement an alternative + // routing peer must show before an otherwise equivalent current peer is + // replaced. An absolute margin alone would be meaningless on slow paths, + // hence latencySwitchGainPercent on top. + latencySwitchMinGain = 20 * time.Millisecond + + // latencySwitchGainPercent is the relative latency improvement required in + // addition to latencySwitchMinGain. A relative margin alone would churn on + // fast paths, where a few milliseconds are a large fraction of the latency. + latencySwitchGainPercent = 20 + + // latencyNoiseSigmas is how many standard deviations of the measurement + // noise a latency gain has to exceed. Smoothing narrows the scatter of a + // jittery path but does not remove it, so on such a path two equivalent + // peers keep producing differences of tens of milliseconds. Requiring the + // gain to stand out from that scatter is what separates a peer that is + // really closer from one that only looks closer right now. + latencyNoiseSigmas = 4 + + // latencySwitchMinDwell is how long a routing peer is kept before latency + // can move the choice again. Changes in connection status, metric or relay + // state ignore it so failover is never delayed. + latencySwitchMinDwell = 30 * time.Second + + // routeReevalInterval is how often routes are re-evaluated without an + // incoming peer state change. A switch deferred by latencySwitchMinDwell, + // or a latency that drifts in steps too small to notify, would otherwise + // wait for an unrelated event to be reconsidered. + routeReevalInterval = latencySwitchMinDwell + + // unknownLatency ranks peers without a latency sample below any peer with a + // real measurement, without excluding them from selection. It doubles as the + // upper bound for samples, so a pathological value cannot dominate, and sits + // well above any usable path so that slow paths keep their order. + unknownLatency = 10 * time.Second + + // holdReportInterval is how long a routing peer held back by a switch rule + // is reported at debug level only, after being reported at info level once. + holdReportInterval = 10 * time.Minute +) + type reason int const ( @@ -36,10 +80,133 @@ const ( reasonHA ) +// switchReason describes why a routing peer replaced the current one. +type switchReason string + +const ( + switchReasonNone switchReason = "" + switchReasonUnavailable switchReason = "current routing peer unavailable" + switchReasonConnected switchReason = "connected peer preferred over idle" + switchReasonMetric switchReason = "lower route metric" + switchReasonDirect switchReason = "direct connection preferred over relayed" + switchReasonLatency switchReason = "lower latency" + switchReasonRouteChanged switchReason = "route assigned to a different routing peer" +) + +// holdKind is the rule that keeps the current routing peer over a better one. +type holdKind int + +const ( + holdLatencyMargin holdKind = iota + holdLatencyDwell + holdRelayedDelay +) + type routerPeerStatus struct { status peer.ConnStatus relayed bool latency time.Duration + // noise is the uncertainty of latency, see peer.LatencySample + noise time.Duration +} + +func (s routerPeerStatus) isDirect() bool { + return s.status == peer.StatusConnected && !s.relayed +} + +// routeCandidate is a route considered for selection, together with the +// selection criteria derived from the state of its routing peer. +type routeCandidate struct { + id route.ID + peer string + metric int + connected bool + relayed bool + latency time.Duration + noise time.Duration +} + +func newRouteCandidate(r *route.Route, status routerPeerStatus) routeCandidate { + latency := status.latency + if latency <= 0 || latency > unknownLatency { + latency = unknownLatency + } + + return routeCandidate{ + id: r.ID, + peer: r.Peer, + metric: r.Metric, + connected: status.status == peer.StatusConnected, + relayed: status.relayed, + latency: latency, + noise: status.noise, + } +} + +// tierCompare orders c against o on the categorical criteria: connection status, +// then route metric, then direct versus relayed. It returns a positive number +// when c outranks o, a negative one when o outranks c, and zero when the two are +// equivalent. Latency is excluded on purpose: it is the tie-break applied by +// betterThan and the only criterion the switch margins in shouldSwitchRoute +// apply to. +func (c routeCandidate) tierCompare(o routeCandidate) int { + if c.connected != o.connected { + if c.connected { + return 1 + } + return -1 + } + if c.metric != o.metric { + return o.metric - c.metric + } + if c.relayed != o.relayed { + if c.relayed { + return -1 + } + return 1 + } + + return 0 +} + +func (c routeCandidate) isDirect() bool { + return c.connected && !c.relayed +} + +func (c routeCandidate) tierBetterThan(o routeCandidate) bool { + return c.tierCompare(o) > 0 +} + +// winsOnlyOnRelay reports whether c outranks o solely by being direct while o +// is relayed, with connection status and metric equal. +func (c routeCandidate) winsOnlyOnRelay(o routeCandidate) bool { + return c.connected == o.connected && c.metric == o.metric && o.relayed && !c.relayed +} + +// switchReasonOver names the criterion on which c replaces the current o. +func (c routeCandidate) switchReasonOver(o routeCandidate) switchReason { + switch { + case c.connected != o.connected: + return switchReasonConnected + case c.metric != o.metric: + return switchReasonMetric + case c.relayed != o.relayed: + return switchReasonDirect + default: + return switchReasonLatency + } +} + +func (c routeCandidate) betterThan(o routeCandidate) bool { + if tier := c.tierCompare(o); tier != 0 { + return tier > 0 + } + + return c.latency < o.latency +} + +func (c routeCandidate) String() string { + return fmt.Sprintf("connected: %t, metric: %d, relayed: %t, latency: %v +/- %v", c.connected, c.metric, c.relayed, c.latency, c.noise) } type RoutesUpdate struct { @@ -69,18 +236,72 @@ type WatcherConfig struct { // Once stopped, it cannot be reused. // The methods are not thread-safe and should be synchronized externally. type Watcher struct { - ctx context.Context - cancel context.CancelFunc - statusRecorder *peer.Status - wgInterface iface.WGIface - routes map[route.ID]*route.Route - routeUpdate chan RoutesUpdate - peerStateUpdate chan map[string]peer.RouterState + ctx context.Context + cancel context.CancelFunc + statusRecorder *peer.Status + wgInterface iface.WGIface + routes map[route.ID]*route.Route + routeUpdate chan RoutesUpdate + // peerStateUpdate holds at most one pending routing peer state notification, + // see watchPeerStatusChanges + peerStateUpdate chan struct{} routePeersNotifiers map[string]chan struct{} // map of peer key to channel for peer state changes currentChosen *route.Route currentChosenStatus *routerPeerStatus - handler RouteHandler - updateSerial uint64 + // lastSwitch is when currentChosen was last replaced by a latency-driven + // switch, used to enforce latencySwitchMinDwell. Categorical switches + // (failover, metric, relay state) leave it untouched: the dwell exists to + // stop latency ping-pong, and arming it on a failover would delay the + // return to a recovered better peer. + lastSwitch time.Time + // latencySwitch marks the evaluation's switch decision as latency-driven; + // set by shouldSwitchRoute, consumed when the switch is applied + latencySwitch bool + // switchReason describes why the evaluation replaced the current routing + // peer; set by getBestRouteFromStatuses, consumed when the switch is applied + switchReason switchReason + // relayedSwitch is how long a current routing peer that dropped from direct + // to relayed is kept over an otherwise equivalent direct one. The zero value + // switches at once. + relayedSwitch relayedSwitchPolicy + // trackedCurrent, trackedPeer and trackedDirect are the current route, its + // routing peer and whether that peer was connected directly as of the + // previous evaluation, used to tell a peer that dropped to relay while + // carrying the route from one that was never direct + trackedCurrent route.ID + trackedPeer string + trackedDirect bool + // degradedSince is when the current routing peer dropped from direct to + // relayed, zero if it is direct or was already relayed when chosen + degradedSince time.Time + // relayedHold fires when a relayed switch being held back becomes due, so it + // is applied on time rather than on the next periodic re-evaluation + relayedHold *time.Timer + // heldCandidate and heldKind identify the alternative last held back by the + // switch rules and the rule holding it, and heldReported is when that was + // reported at info level. Routes are re-evaluated on every latency change, + // so a hold is reported once per episode instead of on every evaluation. + heldCandidate route.ID + heldKind holdKind + heldReported time.Time + // reportedUnassigned tracks that the current no-routing-peer episode has + // been reported, so the periodic re-evaluation does not repeat it + reportedUnassigned bool + // done is closed when the Start loop exits, so Stop can wait for it before + // removing the routes the loop might still be updating + done chan struct{} + // lifecycleMu orders Start against Stop: Start runs in its own goroutine + // and may begin after Stop, or never, so Stop only waits for done when the + // loop has actually started + lifecycleMu sync.Mutex + started bool + stopped bool + handler RouteHandler + updateSerial uint64 +} + +func newPeerStateUpdate() chan struct{} { + return make(chan struct{}, 1) } func NewWatcher(config WatcherConfig) *Watcher { @@ -94,9 +315,11 @@ func NewWatcher(config WatcherConfig) *Watcher { routes: make(map[route.ID]*route.Route), routePeersNotifiers: make(map[string]chan struct{}), routeUpdate: make(chan RoutesUpdate), - peerStateUpdate: make(chan map[string]peer.RouterState), + peerStateUpdate: newPeerStateUpdate(), + done: make(chan struct{}), handler: config.Handler, currentChosenStatus: nil, + relayedSwitch: relayedSwitch(), } return client } @@ -113,23 +336,7 @@ func (w *Watcher) getRouterPeerStatuses() map[route.ID]routerPeerStatus { status: peerStatus.ConnStatus, relayed: peerStatus.Relayed, latency: peerStatus.Latency, - } - } - return routePeerStatuses -} - -func (w *Watcher) convertRouterPeerStatuses(states map[string]peer.RouterState) map[route.ID]routerPeerStatus { - routePeerStatuses := make(map[route.ID]routerPeerStatus) - for _, r := range w.routes { - peerStatus, ok := states[r.Peer] - if !ok { - log.Warnf("couldn't fetch peer state: %v", r.Peer) - continue - } - routePeerStatuses[r.ID] = routerPeerStatus{ - status: peerStatus.Status, - relayed: peerStatus.Relayed, - latency: peerStatus.Latency, + noise: peerStatus.LatencyNoise, } } return routePeerStatuses @@ -139,122 +346,277 @@ func (w *Watcher) convertRouterPeerStatuses(states map[string]peer.RouterState) // within a Watcher, taking into account peer connection status, route metrics, and // preference for non-relayed and direct connections. // -// It follows these prioritization rules: +// Candidates are ranked by the categorical criteria first, in this order: // * Connection status: Both connected and idle peers are considered, but connected peers always take precedence. -// * Idle peer penalty: Idle peers receive a significant score penalty to ensure any connected peer is preferred. // * Metric: Routes with lower metrics (better) are prioritized. // * Non-relayed: Routes without relays are preferred. -// * Latency: Routes with lower latency are prioritized. -// * Allowed IPs: Idle peers can still receive allowed IPs to enable lazy connection triggering. -// * we compare the current score + 10ms to the chosen score to avoid flapping between routes -// * Stability: In case of equal scores, the currently active route (if any) is maintained. +// Peers that are equal on all of them are then ranked by latency, lowest first. +// +// Idle peers can still receive allowed IPs to enable lazy connection triggering. +// Connecting peers are skipped: they have no endpoint to assign allowed IPs to. +// +// Replacing the current route is subject to shouldSwitchRoute, so latency alone +// only moves the route when the gain is worth the interruption. In case of a +// tie, the currently active route (if any) is maintained. // // It returns the ID of the selected optimal route. func (w *Watcher) getBestRouteFromStatuses(routePeerStatuses map[route.ID]routerPeerStatus) (route.ID, routerPeerStatus) { - var chosen route.ID - chosenScore := float64(0) - currScore := float64(0) - var currID route.ID if w.currentChosen != nil { currID = w.currentChosen.ID } - var chosenStatus routerPeerStatus + var best, current routeCandidate + var haveBest, haveCurrent bool for _, r := range w.routes { - tempScore := float64(0) peerStatus, found := routePeerStatuses[r.ID] - // connecting status equals disconnected: no wireguard endpoint to assign allowed IPs to + // connecting status equals disconnected: no endpoint to assign allowed IPs to if !found || peerStatus.status == peer.StatusConnecting { continue } - if r.Metric < route.MaxMetric { - metricDiff := route.MaxMetric - r.Metric - tempScore = float64(metricDiff) * 10 - } - - // in some temporal cases, latency can be 0, so we set it to 999ms to not block but try to avoid this route - latency := 999 * time.Millisecond - if peerStatus.latency != 0 { - latency = peerStatus.latency - } else if !peerStatus.relayed && peerStatus.status != peer.StatusIdle { - log.Tracef("peer %s has 0 latency: [%v]", r.Peer, w.handler) - } - - // avoid negative tempScore on the higher latency calculation - if latency > 1*time.Second { - latency = 999 * time.Millisecond - } - - // higher latency is worse score - tempScore += 1 - latency.Seconds() - - // apply significant penalty for idle peers to ensure connected peers always take precedence - if peerStatus.status == peer.StatusConnected { - tempScore += 100_000 - } - - if !peerStatus.relayed { - tempScore++ - } - - if tempScore > chosenScore || (tempScore == chosenScore && chosen == "") { - chosen = r.ID - chosenStatus = peerStatus - chosenScore = tempScore - } - - if chosen == "" && currID == "" { - chosen = r.ID - chosenStatus = peerStatus - chosenScore = tempScore - } - + candidate := newRouteCandidate(r, peerStatus) if r.ID == currID { - currScore = tempScore + current, haveCurrent = candidate, true + } + if !haveBest || candidate.betterThan(best) { + best, haveBest = candidate, true } } - chosenID := chosen - if chosen == "" { - chosenID = "" - } - currentID := currID - if currID == "" { - currentID = "" + // Candidates that tie must not displace the current peer: the routes are + // held in a map, so the winner of a tie would otherwise depend on iteration + // order and change from one evaluation to the next. + if haveCurrent && !best.betterThan(current) { + best = current } - log.Debugf("chosen route: %s, chosen score: %f, current route: %s, current score: %f", chosenID, chosenScore, currentID, currScore) + w.trackRelayed(current, haveCurrent) + + if !haveBest { + w.reportUnassigned() + w.heldCandidate = "" + + return "", routerPeerStatus{} + } + w.reportedUnassigned = false + + if haveCurrent && current.id != best.id { + if log.IsLevelEnabled(log.DebugLevel) { + log.Debugf("best routing peer for [%v] is %s (%s), current is %s (%s)", + w.handler, best.peer, best, current.peer, current) + } + + if !w.shouldSwitchRoute(current, best) { + return current.id, routePeerStatuses[current.id] + } + } switch { - case chosen == "": - var peers []string - for _, r := range w.routes { - peers = append(peers, r.Peer) - } - - log.Infof("network [%v] has not been assigned a routing peer as no peers from the list %s are currently available", w.handler, peers) - case chosen != currID: - // we compare the current score + 10ms to the chosen score to avoid flapping between routes - if currScore != 0 && currScore+0.01 > chosenScore { - log.Debugf("keeping current routing peer %s for [%v]: the score difference with latency is less than 0.01(10ms): current: %f, new: %f", - w.currentChosen.Peer, w.handler, currScore, chosenScore) - return currID, chosenStatus - } - var p string - if rt := w.routes[chosen]; rt != nil { - p = rt.Peer - } - log.Infof("New chosen route is %s with peer %s with score %f for network [%v]", chosen, p, chosenScore, w.handler) + case haveCurrent && current.id == best.id: + case haveCurrent: + w.switchReason = best.switchReasonOver(current) + log.Warnf("switching routing peer for network [%v] from %s to %s: %s (%s)", + w.handler, current.peer, best.peer, w.switchReason, best) + case w.currentChosen != nil: + w.switchReason = switchReasonUnavailable + log.Warnf("switching routing peer for network [%v] from %s to %s: %s (%s)", + w.handler, w.currentChosen.Peer, best.peer, w.switchReason, best) + default: + log.Infof("new chosen route is %s with peer %s for network [%v]: %s", + best.id, best.peer, w.handler, best) } - return chosen, chosenStatus + return best.id, routePeerStatuses[best.id] } -func (w *Watcher) watchPeerStatusChanges(ctx context.Context, peerKey string, peerStateUpdate chan map[string]peer.RouterState, closer chan struct{}) { - subscription := w.statusRecorder.SubscribeToPeerStateChanges(ctx, peerKey) +// reportUnassigned reports that no routing peer is available for the network, +// once per episode: the periodic re-evaluation would otherwise repeat it for as +// long as the routing peers stay unavailable. +func (w *Watcher) reportUnassigned() { + if w.reportedUnassigned { + return + } + w.reportedUnassigned = true + + var peers []string + for _, r := range w.routes { + peers = append(peers, r.Peer) + } + log.Infof("network [%v] has not been assigned a routing peer as no peers from the list %s are currently available", w.handler, peers) +} + +// trackRelayed records when the current routing peer dropped from a direct to a +// relayed connection while carrying the route, which starts the relayedSwitch +// delay. A peer that was never direct, because it was relayed when chosen or +// came up over relay from idle, is not degraded: holding on to it would only +// keep the route on the slower path. +func (w *Watcher) trackRelayed(current routeCandidate, haveCurrent bool) { + if !haveCurrent { + w.trackedCurrent, w.trackedPeer, w.trackedDirect, w.degradedSince = "", "", false, time.Time{} + return + } + + switch { + case current.id != w.trackedCurrent, current.peer != w.trackedPeer, !current.relayed: + w.degradedSince = time.Time{} + case w.trackedDirect: + w.degradedSince = time.Now() + // a new degradation, reported as such even if an earlier one held + // back the same candidate + w.heldCandidate = "" + } + w.trackedCurrent, w.trackedPeer, w.trackedDirect = current.id, current.peer, current.isDirect() +} + +// shouldSwitchRoute reports whether the current routing peer should be replaced +// by the candidate, which outranks it. +// +// A candidate that wins on a categorical criterion (it is connected while the +// current peer is not, has a better metric, or is direct while the current peer +// is relayed) takes over immediately. The exception is a current peer that only +// just dropped from direct to relayed, see relayedSwitchDue. +// +// A candidate that only has lower latency has to beat the current peer by the +// switch margin, and the current peer has to have been in place for +// latencySwitchMinDwell. Every switch interrupts the traffic using the route, so +// a gain that is marginal, or that the measurement noise of the two paths can +// account for on its own, is not worth acting on. +// +// An approved latency-driven switch sets w.latencySwitch, which arms the dwell +// time once the switch is applied. +func (w *Watcher) shouldSwitchRoute(current, candidate routeCandidate) bool { + if candidate.winsOnlyOnRelay(current) { + return w.relayedSwitchDue(current, candidate) + } + if candidate.tierBetterThan(current) { + return true + } + + gain := current.latency - candidate.latency + if margin := switchMargin(current, candidate); gain < margin { + if level := w.holdLevel(candidate, holdLatencyMargin); log.IsLevelEnabled(level) { + log.StandardLogger().Logf(level, + "keeping routing peer %s for [%v] over %s: latency gain %v on %v is below the %v switch margin (%s)", + current.peer, w.handler, candidate.peer, gain, current.latency, margin, candidate) + } + return false + } + + if dwelled := time.Since(w.lastSwitch); dwelled < latencySwitchMinDwell { + if level := w.holdLevel(candidate, holdLatencyDwell); log.IsLevelEnabled(level) { + log.StandardLogger().Logf(level, + "keeping routing peer %s for [%v] over %s: chosen %v ago, below the %v minimum dwell time", + current.peer, w.handler, candidate.peer, dwelled.Round(time.Second), latencySwitchMinDwell) + } + return false + } + + w.latencySwitch = true + + return true +} + +// relayedSwitchDue reports whether a relayed current routing peer should give +// way to a candidate that is better only by being direct. +// +// A peer that dropped from direct to relayed while carrying the route is kept +// for the relayedSwitch delay: the relay still carries the traffic, staying +// keeps every flow through this routing peer alive, and a direct connection that +// dropped often recovers within seconds. A peer that was already relayed when +// chosen gives way at once, so the route is not stuck on the slower path when a +// direct one becomes available. A switch that is not due yet arms relayedHold so +// it is applied as soon as it becomes due. +func (w *Watcher) relayedSwitchDue(current, candidate routeCandidate) bool { + if w.degradedSince.IsZero() { + return true + } + + policy := w.relayedSwitch + if !policy.never { + remaining := policy.delay - time.Since(w.degradedSince) + if remaining <= 0 { + return true + } + w.armRelayedHold(remaining) + } + + if level := w.holdLevel(candidate, holdRelayedDelay); log.IsLevelEnabled(level) { + log.StandardLogger().Logf(level, + "keeping relayed routing peer %s for [%v] over direct %s: dropped to relay %v ago, switch delay %s", + current.peer, w.handler, candidate.peer, time.Since(w.degradedSince).Round(time.Second), policy) + } + + return false +} + +func (w *Watcher) armRelayedHold(d time.Duration) { + if w.relayedHold == nil { + w.relayedHold = time.NewTimer(d) + return + } + w.relayedHold.Reset(d) +} + +// relayedHoldC is the channel of relayedHold, nil while it was never armed so +// that selecting on it blocks. +func (w *Watcher) relayedHoldC() <-chan time.Time { + if w.relayedHold == nil { + return nil + } + return w.relayedHold.C +} + +// holdLevel is the level to report a decision to keep the current routing peer +// at. The first evaluation that holds a given candidate back under a given rule +// is reported at info level, because that is what someone wondering why a route +// did not move needs to see. The repeats stay at debug level for +// holdReportInterval, including when two near-equal peers keep trading places. +func (w *Watcher) holdLevel(candidate routeCandidate, kind holdKind) log.Level { + if w.heldCandidate == candidate.id && w.heldKind == kind && time.Since(w.heldReported) < holdReportInterval { + return log.DebugLevel + } + + w.heldCandidate, w.heldKind, w.heldReported = candidate.id, kind, time.Now() + + return log.InfoLevel +} + +// switchMargin is how much lower the candidate's latency has to be before the +// route moves: at least latencySwitchMinGain, at least latencySwitchGainPercent +// of the current latency so the margin scales with slow paths, and more than the +// combined measurement noise of the two paths. The noise term is what keeps a +// jittery path from moving the route on its own; it vanishes on a stable path, +// where the fixed margins decide. +func switchMargin(current, candidate routeCandidate) time.Duration { + margin := latencySwitchMinGain + + if relative := current.latency * latencySwitchGainPercent / 100; relative > margin { + margin = relative + } + + // the two paths vary independently, so their noise adds in quadrature + noise := time.Duration(latencyNoiseSigmas * math.Hypot(float64(current.noise), float64(candidate.noise))) + if noise > margin { + margin = noise + } + + return margin +} + +// watchPeerStatusChanges forwards the state changes of a routing peer to the +// watcher as a notification to re-evaluate. The watcher reads the current peer +// states when it gets to it, so a notification still pending covers any number +// of later changes, and the forwarder never waits for a busy watcher: that +// would stop it draining the subscription and in turn block the peer's updates. +func (w *Watcher) watchPeerStatusChanges(ctx context.Context, peerKey string, peerStateUpdate chan<- struct{}, closer chan struct{}) { + // the subscription gets its own context so a delivery blocked on it is + // released as soon as this forwarder exits, not only when the watcher stops + subCtx, cancel := context.WithCancel(ctx) + defer cancel() + + subscription := w.statusRecorder.SubscribeToPeerStateChanges(subCtx, peerKey) defer w.statusRecorder.UnsubscribePeerStateChanges(subscription) for { @@ -263,14 +625,11 @@ func (w *Watcher) watchPeerStatusChanges(ctx context.Context, peerKey string, pe return case <-closer: return - case routerStates := <-subscription.Events(): + case <-subscription.Events(): select { - case peerStateUpdate <- routerStates: + case peerStateUpdate <- struct{}{}: log.Debugf("triggered route state update for Peer: %s", peerKey) - case <-ctx.Done(): - return - case <-closer: - return + default: } } } @@ -337,6 +696,8 @@ func (w *Watcher) shouldSkipRecalculation(newChosenID route.ID, newStatus router } func (w *Watcher) recalculateRoutes(rsn reason, routerPeerStatuses map[route.ID]routerPeerStatus) error { + w.latencySwitch = false + w.switchReason = switchReasonNone newChosenID, newStatus := w.getBestRouteFromStatuses(routerPeerStatuses) // If no route is chosen, remove the route from the peer @@ -348,6 +709,9 @@ func (w *Watcher) recalculateRoutes(rsn reason, routerPeerStatuses map[route.ID] if err := w.removeAllowedIPs(w.currentChosen, rsn); err != nil { return fmt.Errorf("remove obsolete: %w", err) } + if rsn == reasonPeerUpdate { + w.routingPeerLostEvent(w.currentChosen) + } w.currentChosen = nil w.currentChosenStatus = nil @@ -360,6 +724,17 @@ func (w *Watcher) recalculateRoutes(rsn reason, routerPeerStatuses map[route.ID] return nil } + previous := w.currentChosen + newChosenRoute := w.routes[newChosenID] + // a route update can hand the same route to a different routing peer + peerReplaced := previous != nil && previous.ID == newChosenID && previous.Peer != newChosenRoute.Peer + switched := previous == nil || previous.ID != newChosenID || peerReplaced + if peerReplaced { + w.switchReason = switchReasonRouteChanged + log.Warnf("switching routing peer for network [%v] from %s to %s: %s", + w.handler, previous.Peer, newChosenRoute.Peer, w.switchReason) + } + // If the chosen route was assigned to a different peer, remove the allowed IPs first if isNew := w.currentChosen == nil; !isNew { if err := w.removeAllowedIPs(w.currentChosen, reasonHA); err != nil { @@ -367,7 +742,6 @@ func (w *Watcher) recalculateRoutes(rsn reason, routerPeerStatuses map[route.ID] } } - newChosenRoute := w.routes[newChosenID] if err := w.addAllowedIPs(newChosenRoute); err != nil { return fmt.Errorf("add new: %w", err) } @@ -377,20 +751,86 @@ func (w *Watcher) recalculateRoutes(rsn reason, routerPeerStatuses map[route.ID] w.currentChosen = newChosenRoute w.currentChosenStatus = &newStatus + if !switched { + return nil + } + + // start tracking the new routing peer from the state it was chosen in, so a + // drop to relay seen by the next evaluation counts as a degradation + w.trackedCurrent, w.trackedPeer = newChosenID, newChosenRoute.Peer + w.trackedDirect, w.degradedSince = newStatus.isDirect(), time.Time{} + if w.latencySwitch { + w.lastSwitch = time.Now() + } + if previous != nil { + w.routingPeerSwitchEvent(previous, newChosenRoute) + } return nil } -func (w *Watcher) connectEvent(route *route.Route) { - var defaultRoute bool - for _, r := range w.routes { - if r.Network.Bits() == 0 { - defaultRoute = true - break - } +// routingPeerSwitchEvent publishes a replaced routing peer as a system event, +// so it shows in the status output and debug bundles whatever the log level. +// Moving a masqueraded route resets the connections that used it. For default +// routes this adds the reason to the events of disconnectEvent and +// connectEvent, and carries no user message so it is not shown as a +// notification a second time. +func (w *Watcher) routingPeerSwitchEvent(previous, next *route.Route) { + if w.statusRecorder == nil { + return } - if !defaultRoute { + reason := w.switchReason + if reason == switchReasonNone { + reason = switchReasonUnavailable + } + + w.statusRecorder.PublishEvent( + proto.SystemEvent_WARNING, + proto.SystemEvent_NETWORK, + "Routing peer changed", + "", + map[string]string{ + "network": w.handler.String(), + "id": string(next.NetID), + "previous_peer": previous.Peer, + "peer": next.Peer, + "reason": string(reason), + }, + ) +} + +// routingPeerLostEvent publishes the loss of the last available routing peer of +// a network as a system event. Default routes are left to disconnectEvent. +func (w *Watcher) routingPeerLostEvent(previous *route.Route) { + if w.statusRecorder == nil || w.hasDefaultRoute() { + return + } + + w.statusRecorder.PublishEvent( + proto.SystemEvent_WARNING, + proto.SystemEvent_NETWORK, + "No routing peer available", + "", + map[string]string{ + "network": w.handler.String(), + "id": string(previous.NetID), + "previous_peer": previous.Peer, + }, + ) +} + +func (w *Watcher) hasDefaultRoute() bool { + for _, r := range w.routes { + if r.Network.Bits() == 0 { + return true + } + } + return false +} + +func (w *Watcher) connectEvent(route *route.Route) { + if !w.hasDefaultRoute() { return } @@ -411,15 +851,7 @@ func (w *Watcher) connectEvent(route *route.Route) { } func (w *Watcher) disconnectEvent(route *route.Route, rsn reason) { - var defaultRoute bool - for _, r := range w.routes { - if r.Network.Bits() == 0 { - defaultRoute = true - break - } - } - - if !defaultRoute { + if !w.hasDefaultRoute() { return } @@ -505,13 +937,41 @@ func (w *Watcher) classifyUpdate(update RoutesUpdate) bool { // Start is the main point of reacting on client network routing events. // All the processing related to the client network should be done here. Thread-safe. func (w *Watcher) Start() { + w.lifecycleMu.Lock() + if w.stopped || w.started { + w.lifecycleMu.Unlock() + return + } + w.started = true + w.lifecycleMu.Unlock() + + defer close(w.done) + + reeval := time.NewTicker(routeReevalInterval) + defer reeval.Stop() + defer func() { + if w.relayedHold != nil { + w.relayedHold.Stop() + } + }() + for { select { case <-w.ctx.Done(): return - case routersStates := <-w.peerStateUpdate: - routerPeerStatuses := w.convertRouterPeerStatuses(routersStates) - if err := w.recalculateRoutes(reasonPeerUpdate, routerPeerStatuses); err != nil { + case <-reeval.C: + if err := w.recalculateRoutes(reasonPeerUpdate, w.getRouterPeerStatuses()); err != nil { + log.Errorf("Failed to re-evaluate routes for network [%v]: %v", w.handler, err) + } + case <-w.relayedHoldC(): + if err := w.recalculateRoutes(reasonPeerUpdate, w.getRouterPeerStatuses()); err != nil { + log.Errorf("Failed to re-evaluate routes for network [%v]: %v", w.handler, err) + } + case <-w.peerStateUpdate: + // The notification only says that a routing peer changed. The states it + // carries can be older than ones already applied, since every routing + // peer has its own subscription, so the current states are read instead. + if err := w.recalculateRoutes(reasonPeerUpdate, w.getRouterPeerStatuses()); err != nil { log.Errorf("Failed to recalculate routes for network [%v]: %v", w.handler, err) } case update := <-w.routeUpdate: @@ -550,7 +1010,22 @@ func (w *Watcher) handleRouteUpdate(update RoutesUpdate) { func (w *Watcher) Stop() { log.Debugf("Stopping watcher for network [%v]", w.handler) + w.lifecycleMu.Lock() + if w.stopped { + w.lifecycleMu.Unlock() + return + } + w.stopped = true + started := w.started + w.lifecycleMu.Unlock() + w.cancel() + // wait for the Start loop to exit: it may be inside a recalculation, and + // cleaning up concurrently would race on currentChosen and could leave the + // allowed IPs of a route chosen after this point behind + if started { + <-w.done + } if w.currentChosen == nil { return diff --git a/client/internal/routemanager/client/client_bench_test.go b/client/internal/routemanager/client/client_bench_test.go index 668aec427..b788015db 100644 --- a/client/internal/routemanager/client/client_bench_test.go +++ b/client/internal/routemanager/client/client_bench_test.go @@ -23,6 +23,12 @@ var benchmarkTiers = []benchmarkTier{ {"Small", 100, 50, 4}, {"Medium", 1000, 200, 16}, {"Large", 5000, 500, 32}, + // the tiers below are deliberately past anything realistic: a watcher holds + // the routes of a single network, while these give one watcher every route + // in the account at once + {"ManyPeers", 20000, 500, 32}, + {"ManyRoutes", 1000, 5000, 16}, + {"ManyBoth", 20000, 5000, 32}, } type mockRouteHandler struct { @@ -136,7 +142,7 @@ func BenchmarkRecalculateRoutes(b *testing.B) { routes: routes, routePeersNotifiers: make(map[string]chan struct{}), routeUpdate: make(chan RoutesUpdate), - peerStateUpdate: make(chan map[string]peer.RouterState), + peerStateUpdate: newPeerStateUpdate(), handler: &mockRouteHandler{network: "benchmark"}, currentChosenStatus: nil, } diff --git a/client/internal/routemanager/client/client_test.go b/client/internal/routemanager/client/client_test.go index 850f6691f..764a2f74d 100644 --- a/client/internal/routemanager/client/client_test.go +++ b/client/internal/routemanager/client/client_test.go @@ -1,11 +1,17 @@ package client import ( + "context" "fmt" "net/netip" + "sync" "testing" "time" + log "github.com/sirupsen/logrus" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/netbirdio/netbird/client/internal/peer" "github.com/netbirdio/netbird/client/internal/routemanager/common" "github.com/netbirdio/netbird/client/internal/routemanager/static" @@ -316,6 +322,145 @@ func TestGetBestrouteFromStatuses(t *testing.T) { currentRoute: "route1", expectedRouteID: "route2", }, + { + // 40ms gain clears latencySwitchMinGain but not the 20% of 300ms + name: "latency gain below the relative margin should not change the route", + statuses: map[route.ID]routerPeerStatus{ + "route1": { + status: peer.StatusConnected, + latency: 300 * time.Millisecond, + }, + "route2": { + status: peer.StatusConnected, + latency: 260 * time.Millisecond, + }, + }, + existingRoutes: map[route.ID]*route.Route{ + "route1": { + ID: "route1", + Metric: route.MaxMetric, + Peer: "peer1", + }, + "route2": { + ID: "route2", + Metric: route.MaxMetric, + Peer: "peer2", + }, + }, + currentRoute: "route1", + expectedRouteID: "route1", + }, + { + name: "latency gain above both margins should change the route", + statuses: map[route.ID]routerPeerStatus{ + "route1": { + status: peer.StatusConnected, + latency: 300 * time.Millisecond, + }, + "route2": { + status: peer.StatusConnected, + latency: 200 * time.Millisecond, + }, + }, + existingRoutes: map[route.ID]*route.Route{ + "route1": { + ID: "route1", + Metric: route.MaxMetric, + Peer: "peer1", + }, + "route2": { + ID: "route2", + Metric: route.MaxMetric, + Peer: "peer2", + }, + }, + currentRoute: "route1", + expectedRouteID: "route2", + }, + { + // 7ms gain is 58% of the current latency but below latencySwitchMinGain + name: "latency gain below the absolute margin should not change the route", + statuses: map[route.ID]routerPeerStatus{ + "route1": { + status: peer.StatusConnected, + latency: 12 * time.Millisecond, + }, + "route2": { + status: peer.StatusConnected, + latency: 5 * time.Millisecond, + }, + }, + existingRoutes: map[route.ID]*route.Route{ + "route1": { + ID: "route1", + Metric: route.MaxMetric, + Peer: "peer1", + }, + "route2": { + ID: "route2", + Metric: route.MaxMetric, + Peer: "peer2", + }, + }, + currentRoute: "route1", + expectedRouteID: "route1", + }, + { + name: "current relayed peer should be replaced by a direct peer with worse latency", + statuses: map[route.ID]routerPeerStatus{ + "route1": { + status: peer.StatusConnected, + relayed: true, + latency: 10 * time.Millisecond, + }, + "route2": { + status: peer.StatusConnected, + relayed: false, + latency: 200 * time.Millisecond, + }, + }, + existingRoutes: map[route.ID]*route.Route{ + "route1": { + ID: "route1", + Metric: route.MaxMetric, + Peer: "peer1", + }, + "route2": { + ID: "route2", + Metric: route.MaxMetric, + Peer: "peer2", + }, + }, + currentRoute: "route1", + expectedRouteID: "route2", + }, + { + name: "better metric should replace the current peer with better latency", + statuses: map[route.ID]routerPeerStatus{ + "route1": { + status: peer.StatusConnected, + latency: 10 * time.Millisecond, + }, + "route2": { + status: peer.StatusConnected, + latency: 300 * time.Millisecond, + }, + }, + existingRoutes: map[route.ID]*route.Route{ + "route1": { + ID: "route1", + Metric: route.MaxMetric, + Peer: "peer1", + }, + "route2": { + ID: "route2", + Metric: 5000, + Peer: "peer2", + }, + }, + currentRoute: "route1", + expectedRouteID: "route2", + }, { name: "current chosen route doesn't exist anymore", statuses: map[route.ID]routerPeerStatus{ @@ -828,3 +973,510 @@ func TestGetBestrouteFromStatuses(t *testing.T) { }) } } + +// newTestWatcher builds a watcher over two routes of equal metric, with route1 +// as the currently chosen one. +func newTestWatcher(t *testing.T, lastSwitch time.Time) *Watcher { + t.Helper() + + routes := map[route.ID]*route.Route{ + "route1": {ID: "route1", Metric: route.MaxMetric, Peer: "peer1"}, + "route2": {ID: "route2", Metric: route.MaxMetric, Peer: "peer2"}, + } + + params := common.HandlerParams{ + Route: &route.Route{Network: netip.MustParsePrefix("192.168.0.0/24")}, + } + + return &Watcher{ + handler: static.NewRoute(params), + routes: routes, + currentChosen: routes["route1"], + lastSwitch: lastSwitch, + } +} + +func TestLatencySwitchDwellTime(t *testing.T) { + // a gain large enough to clear both switch margins + statuses := map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected, latency: 300 * time.Millisecond}, + "route2": {status: peer.StatusConnected, latency: 100 * time.Millisecond}, + } + + t.Run("latency switch is held during the dwell time", func(t *testing.T) { + w := newTestWatcher(t, time.Now()) + + chosen, _ := w.getBestRouteFromStatuses(statuses) + assert.Equal(t, route.ID("route1"), chosen, "route should be kept while the dwell time has not elapsed") + }) + + t.Run("latency switch is allowed after the dwell time", func(t *testing.T) { + w := newTestWatcher(t, time.Now().Add(-latencySwitchMinDwell-time.Second)) + + chosen, _ := w.getBestRouteFromStatuses(statuses) + assert.Equal(t, route.ID("route2"), chosen, "route should switch once the dwell time has elapsed") + }) + + t.Run("categorical improvement ignores the dwell time", func(t *testing.T) { + w := newTestWatcher(t, time.Now()) + + // route1 is no longer connected, so the dwell time must not delay failover + chosen, _ := w.getBestRouteFromStatuses(map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusIdle, latency: 10 * time.Millisecond}, + "route2": {status: peer.StatusConnected, latency: 300 * time.Millisecond}, + }) + assert.Equal(t, route.ID("route2"), chosen, "failover must not be delayed by the dwell time") + }) +} + +// TestLatencySwitchNoiseMargin covers the noise aware margin: on a jittery path +// the smoothed latencies of two equivalent peers keep drifting apart by tens of +// milliseconds, and that difference must not move the route even though it +// clears the fixed margins. +// Measurements on slow paths must keep their order: a peer measured at 1.2s is +// better than one at 2.5s, and a peer without a sample still ranks below both. +func TestSlowPathLatenciesKeepTheirOrder(t *testing.T) { + w := newTestWatcher(t, time.Time{}) + + chosen, _ := w.getBestRouteFromStatuses(map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected, latency: 2500 * time.Millisecond}, + "route2": {status: peer.StatusConnected, latency: 1200 * time.Millisecond}, + }) + assert.Equal(t, route.ID("route2"), chosen, "a 1.3s gain on a 2.5s path must move the route") + + unsampled := newRouteCandidate(w.routes["route1"], routerPeerStatus{status: peer.StatusConnected}) + slow := newRouteCandidate(w.routes["route2"], routerPeerStatus{status: peer.StatusConnected, latency: 5 * time.Second}) + assert.True(t, slow.betterThan(unsampled), "a measured peer must rank above one without a sample") +} + +func TestLatencySwitchNoiseMargin(t *testing.T) { + t.Run("a gain within the measurement noise does not switch", func(t *testing.T) { + w := newTestWatcher(t, time.Time{}) + + chosen, _ := w.getBestRouteFromStatuses(map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected, latency: 120 * time.Millisecond, noise: 12 * time.Millisecond}, + "route2": {status: peer.StatusConnected, latency: 80 * time.Millisecond, noise: 12 * time.Millisecond}, + }) + assert.Equal(t, route.ID("route1"), chosen, "a 40ms gain between two paths jittering by 12ms is not a real difference") + }) + + t.Run("a gain beyond the measurement noise switches", func(t *testing.T) { + w := newTestWatcher(t, time.Time{}) + + chosen, _ := w.getBestRouteFromStatuses(map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected, latency: 300 * time.Millisecond, noise: 12 * time.Millisecond}, + "route2": {status: peer.StatusConnected, latency: 80 * time.Millisecond, noise: 12 * time.Millisecond}, + }) + assert.Equal(t, route.ID("route2"), chosen, "a 220ms gain stands out from 12ms of jitter") + }) + + // the held candidate is tracked so the decision is reported once per episode + // rather than on every re-evaluation + t.Run("holding a candidate back is tracked as one episode", func(t *testing.T) { + w := newTestWatcher(t, time.Time{}) + + within := map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected, latency: 120 * time.Millisecond, noise: 12 * time.Millisecond}, + "route2": {status: peer.StatusConnected, latency: 80 * time.Millisecond, noise: 12 * time.Millisecond}, + } + + w.getBestRouteFromStatuses(within) + assert.Equal(t, route.ID("route2"), w.heldCandidate, "the held candidate should be recorded on the first hold") + + w.getBestRouteFromStatuses(within) + assert.Equal(t, route.ID("route2"), w.heldCandidate, "a repeated hold of the same candidate stays the same episode") + + // the smoothed latencies of two near-equal peers keep crossing: route2 + // falling behind for a moment must not start a new episode + w.getBestRouteFromStatuses(map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected, latency: 80 * time.Millisecond, noise: 12 * time.Millisecond}, + "route2": {status: peer.StatusConnected, latency: 120 * time.Millisecond, noise: 12 * time.Millisecond}, + }) + held := routeCandidate{id: "route2"} + assert.Equal(t, log.DebugLevel, w.holdLevel(held, holdLatencyMargin), "holding route2 back again after a crossing is a repeat") + + // a hold reported long ago is worth reporting again + w.heldReported = w.heldReported.Add(-holdReportInterval - time.Second) + assert.Equal(t, log.InfoLevel, w.holdLevel(held, holdLatencyMargin), "a hold is reported again once the report interval passed") + }) + + t.Run("a stable path is decided by the fixed margins", func(t *testing.T) { + w := newTestWatcher(t, time.Time{}) + + // the same 40ms gain as the first case, on paths that barely vary + chosen, _ := w.getBestRouteFromStatuses(map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected, latency: 120 * time.Millisecond, noise: 300 * time.Microsecond}, + "route2": {status: peer.StatusConnected, latency: 80 * time.Millisecond, noise: 300 * time.Microsecond}, + }) + assert.Equal(t, route.ID("route2"), chosen, "without jitter a 40ms gain on a 120ms path is a real improvement") + }) +} + +// recordingHandler records the allowed IPs calls a watcher makes. +type recordingHandler struct { + added []string + removed int +} + +func (h *recordingHandler) String() string { return "recording" } +func (h *recordingHandler) AddRoute(context.Context) error { return nil } +func (h *recordingHandler) RemoveRoute() error { return nil } +func (h *recordingHandler) AddAllowedIPs(peerKey string) error { + h.added = append(h.added, peerKey) + return nil +} + +func (h *recordingHandler) RemoveAllowedIPs() error { + h.removed++ + return nil +} + +func newRecalcWatcher(t *testing.T) (*Watcher, *recordingHandler) { + t.Helper() + + handler := &recordingHandler{} + routes := map[route.ID]*route.Route{ + "route1": {ID: "route1", Metric: route.MaxMetric, Peer: "peer1"}, + "route2": {ID: "route2", Metric: route.MaxMetric, Peer: "peer2"}, + } + + return &Watcher{ + statusRecorder: peer.NewRecorder("https://mgm"), + routes: routes, + handler: handler, + }, handler +} + +// TestRecalculateRoutesAssignsIdlePeer covers the lazy connection path: an idle +// routing peer has no established connection yet, but must still receive the +// allowed IPs so that traffic for the route can trigger the connection. +func TestRecalculateRoutesAssignsIdlePeer(t *testing.T) { + w, handler := newRecalcWatcher(t) + + err := w.recalculateRoutes(reasonPeerUpdate, map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusIdle}, + }) + require.NoError(t, err) + + assert.Equal(t, []string{"peer1"}, handler.added, "an idle routing peer should still get the allowed IPs") + require.NotNil(t, w.currentChosen, "an idle peer should be chosen when it is the only candidate") + assert.Equal(t, route.ID("route1"), w.currentChosen.ID) +} + +func TestRecalculateRoutesDwellTime(t *testing.T) { + w, handler := newRecalcWatcher(t) + + // route2 has no status yet, so route1 is the only candidate + require.NoError(t, w.recalculateRoutes(reasonPeerUpdate, map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected, latency: 300 * time.Millisecond}, + })) + require.Equal(t, []string{"peer1"}, handler.added) + require.True(t, w.lastSwitch.IsZero(), "the initial assignment is not latency-driven and must not arm the dwell time") + + betterRoute2 := map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected, latency: 300 * time.Millisecond}, + "route2": {status: peer.StatusConnected, latency: 100 * time.Millisecond}, + } + + require.NoError(t, w.recalculateRoutes(reasonPeerUpdate, betterRoute2)) + assert.Equal(t, []string{"peer1", "peer2"}, handler.added, "a latency switch with no recent switch should be allowed") + assert.Equal(t, 1, handler.removed) + require.False(t, w.lastSwitch.IsZero(), "a latency-driven switch must arm the dwell time") + + // the peers trade places: switching back is again latency-only, but now + // inside the dwell window + betterRoute1 := map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected, latency: 100 * time.Millisecond}, + "route2": {status: peer.StatusConnected, latency: 300 * time.Millisecond}, + } + + require.NoError(t, w.recalculateRoutes(reasonPeerUpdate, betterRoute1)) + assert.Equal(t, []string{"peer1", "peer2"}, handler.added, "the route should be held during the dwell time") + + w.lastSwitch = w.lastSwitch.Add(-latencySwitchMinDwell - time.Second) + + require.NoError(t, w.recalculateRoutes(reasonPeerUpdate, betterRoute1)) + assert.Equal(t, []string{"peer1", "peer2", "peer1"}, handler.added, "the route should switch once the dwell time elapsed") + assert.Equal(t, 2, handler.removed) + + // the switch re-armed the dwell time, which must not delay failover + require.NoError(t, w.recalculateRoutes(reasonPeerUpdate, map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnecting, latency: 100 * time.Millisecond}, + "route2": {status: peer.StatusConnected, latency: 300 * time.Millisecond}, + })) + assert.Equal(t, []string{"peer1", "peer2", "peer1", "peer2"}, handler.added, "losing the routing peer must fail over regardless of the dwell time") + assert.Equal(t, 3, handler.removed) +} + +// TestFailoverDoesNotArmDwellTime covers the recovery path: a failover to a +// worse peer must not start the dwell time, or traffic would be stuck on the +// worse path for the full dwell after the better peer comes back. +func TestFailoverDoesNotArmDwellTime(t *testing.T) { + w, handler := newRecalcWatcher(t) + + require.NoError(t, w.recalculateRoutes(reasonPeerUpdate, map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected, latency: 20 * time.Millisecond}, + "route2": {status: peer.StatusConnected, latency: 200 * time.Millisecond}, + })) + require.Equal(t, []string{"peer1"}, handler.added) + + // peer1 blips into connecting: unguarded failover to the worse peer2 + require.NoError(t, w.recalculateRoutes(reasonPeerUpdate, map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnecting, latency: 20 * time.Millisecond}, + "route2": {status: peer.StatusConnected, latency: 200 * time.Millisecond}, + })) + require.Equal(t, []string{"peer1", "peer2"}, handler.added, "failover to the remaining peer should happen") + require.True(t, w.lastSwitch.IsZero(), "a failover must not arm the dwell time") + + // peer1 recovers seconds later: the return is latency-only but must not be + // held back by a dwell the failover would have armed + require.NoError(t, w.recalculateRoutes(reasonPeerUpdate, map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected, latency: 20 * time.Millisecond}, + "route2": {status: peer.StatusConnected, latency: 200 * time.Millisecond}, + })) + assert.Equal(t, []string{"peer1", "peer2", "peer1"}, handler.added, "the recovered better peer should take the route back immediately") +} + +// blockingHandler records the allowed IPs calls in order and blocks the first +// AddAllowedIPs until released, to hold a recalculation in flight. +type blockingHandler struct { + mu sync.Mutex + calls []string + entered chan struct{} + release chan struct{} + once sync.Once +} + +func (h *blockingHandler) String() string { return "blocking" } +func (h *blockingHandler) AddRoute(context.Context) error { return nil } +func (h *blockingHandler) RemoveRoute() error { return nil } + +func (h *blockingHandler) AddAllowedIPs(peerKey string) error { + h.once.Do(func() { + close(h.entered) + <-h.release + }) + h.record("add " + peerKey) + return nil +} + +func (h *blockingHandler) RemoveAllowedIPs() error { + h.record("remove") + return nil +} + +func (h *blockingHandler) record(call string) { + h.mu.Lock() + defer h.mu.Unlock() + h.calls = append(h.calls, call) +} + +// TestStopWaitsForStart covers the shutdown ordering: Stop must not clean up +// while the Start loop is still inside a recalculation, or it would remove the +// allowed IPs before the recalculation adds them and leave them behind. +func TestStopWaitsForStart(t *testing.T) { + w, _ := newRecalcWatcher(t) + handler := &blockingHandler{entered: make(chan struct{}), release: make(chan struct{})} + w.handler = handler + + require.NoError(t, w.statusRecorder.AddPeer("peer1", "peer1.netbird.cloud", "100.64.0.1", "")) + require.NoError(t, w.statusRecorder.UpdatePeerState(peer.State{PubKey: "peer1", ConnStatus: peer.StatusConnected})) + + w.ctx, w.cancel = context.WithCancel(context.Background()) + w.done = make(chan struct{}) + w.peerStateUpdate = make(chan struct{}) + go w.Start() + + // the unbuffered send returns once the loop took the notification, so the + // loop is running and about to recalculate + w.peerStateUpdate <- struct{}{} + select { + case <-handler.entered: + case <-time.After(2 * time.Second): + t.Fatal("the notification should start a recalculation") + } + + finished := make(chan struct{}) + go func() { + w.Stop() + close(finished) + }() + + select { + case <-finished: + t.Fatal("Stop must wait for the recalculation in flight") + case <-time.After(100 * time.Millisecond): + } + + close(handler.release) + select { + case <-finished: + case <-time.After(2 * time.Second): + t.Fatal("Stop should return once the Start loop has exited") + } + + handler.mu.Lock() + defer handler.mu.Unlock() + assert.Equal(t, []string{"add peer1", "remove"}, handler.calls, "Stop must clean up after the recalculation, not before it") +} + +// A notification carries no peer states: every routing peer has its own +// subscription and forwarder feeding the same channel, so a snapshot could +// arrive after a newer one. The watcher acts on the states in the recorder as +// of when it handles the notification. +func TestPeerStateNotificationUsesRecordedStates(t *testing.T) { + w, handler := newRecalcWatcher(t) + + recorder := w.statusRecorder + for _, key := range []string{"peer1", "peer2"} { + require.NoError(t, recorder.AddPeer(key, key+".netbird.cloud", "100.64.0.1", "")) + } + require.NoError(t, recorder.UpdatePeerICEState(peer.State{PubKey: "peer1", ConnStatus: peer.StatusConnected})) + require.NoError(t, recorder.UpdatePeerICEState(peer.State{PubKey: "peer2", ConnStatus: peer.StatusConnecting})) + + require.NoError(t, w.recalculateRoutes(reasonPeerUpdate, w.getRouterPeerStatuses())) + require.Equal(t, []string{"peer1"}, handler.added) + + w.ctx, w.cancel = context.WithCancel(context.Background()) + w.done = make(chan struct{}) + w.peerStateUpdate = make(chan struct{}) + go w.Start() + + require.NoError(t, recorder.UpdatePeerICEState(peer.State{PubKey: "peer1", ConnStatus: peer.StatusConnecting})) + require.NoError(t, recorder.UpdatePeerICEState(peer.State{PubKey: "peer2", ConnStatus: peer.StatusConnected})) + w.peerStateUpdate <- struct{}{} + w.Stop() + + assert.Equal(t, []string{"peer1", "peer2"}, handler.added, "the notification must move the route according to the recorded states") +} + +// While the watcher is busy, for instance inside a recalculation, the peer state +// forwarder must keep draining its subscription. Otherwise the subscription +// buffer fills up and the peer's next state update blocks its caller, the +// peer's connection handling, until the watcher is free again. +func TestPeerStateForwarderDoesNotBlockWhileWatcherIsBusy(t *testing.T) { + w, _ := newRecalcWatcher(t) + w.peerStateUpdate = newPeerStateUpdate() + + recorder := w.statusRecorder + require.NoError(t, recorder.AddPeer("peer1", "peer1.netbird.cloud", "100.64.0.1", "")) + + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + closer := make(chan struct{}) + go w.watchPeerStatusChanges(ctx, "peer1", w.peerStateUpdate, closer) + + // wait for the forwarder's subscription to exist, so the updates below reach it + require.Eventually(t, func() bool { + require.NoError(t, recorder.UpdatePeerICEState(peer.State{PubKey: "peer1", ConnStatus: peer.StatusConnected})) + require.NoError(t, recorder.UpdatePeerICEState(peer.State{PubKey: "peer1", ConnStatus: peer.StatusConnecting})) + select { + case <-w.peerStateUpdate: + return true + case <-time.After(50 * time.Millisecond): + return false + } + }, 2*time.Second, 10*time.Millisecond, "the forwarder must pass on a notification") + + // nobody reads peerStateUpdate: the watcher is busy + updated := make(chan struct{}) + go func() { + defer close(updated) + for i := 0; i < 50; i++ { + status := peer.StatusConnected + if i%2 == 1 { + status = peer.StatusConnecting + } + if err := recorder.UpdatePeerICEState(peer.State{PubKey: "peer1", ConnStatus: status}); err != nil { + t.Errorf("update peer state: %v", err) + return + } + } + }() + + select { + case <-updated: + case <-time.After(2 * time.Second): + t.Fatal("peer state updates must not block while the watcher is busy") + } +} + +// TestStopWithoutStart covers a watcher whose Start goroutine never ran, or ran +// only after Stop: Stop must return instead of waiting for a loop that does not +// exist, and a late Start must not begin working on a stopped watcher. +func TestStopWithoutStart(t *testing.T) { + w := NewWatcher(WatcherConfig{ + Context: context.Background(), + StatusRecorder: peer.NewRecorder("https://mgm"), + Handler: &recordingHandler{}, + }) + + finished := make(chan struct{}) + go func() { + w.Stop() + close(finished) + }() + + select { + case <-finished: + case <-time.After(2 * time.Second): + t.Fatal("Stop must return when Start was never called") + } + + started := make(chan struct{}) + go func() { + w.Start() + close(started) + }() + + select { + case <-started: + case <-time.After(2 * time.Second): + t.Fatal("Start after Stop must return immediately") + } +} + +// TestUnassignedNetworkReportedOncePerEpisode covers the no-routing-peer log: +// the periodic re-evaluation hits the same branch every tick, so the episode +// must be reported once, and again only after a route was assigned in between. +func TestUnassignedNetworkReportedOncePerEpisode(t *testing.T) { + w, _ := newRecalcWatcher(t) + + unassigned := map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnecting}, + } + assigned := map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected, latency: 10 * time.Millisecond}, + } + + require.NoError(t, w.recalculateRoutes(reasonPeerUpdate, unassigned)) + assert.True(t, w.reportedUnassigned, "the first unassigned evaluation should be reported") + + require.NoError(t, w.recalculateRoutes(reasonPeerUpdate, unassigned)) + assert.True(t, w.reportedUnassigned, "repeats stay within the same episode") + + require.NoError(t, w.recalculateRoutes(reasonPeerUpdate, assigned)) + assert.False(t, w.reportedUnassigned, "an assigned route ends the episode") + + require.NoError(t, w.recalculateRoutes(reasonPeerUpdate, unassigned)) + assert.True(t, w.reportedUnassigned, "losing the route again starts a new episode") +} + +// TestLatencyJitterDoesNotSwitchRoute walks two equivalent peers through jitter +// around the same mean. Neither peer is genuinely better, so the route must not +// move at any point. +func TestLatencyJitterDoesNotSwitchRoute(t *testing.T) { + w := newTestWatcher(t, time.Time{}) + + jitter := []int{-6, 3, -4, 8, -2, 5, -7, 1, 6, -3} + for i, offsetMs := range jitter { + offset := time.Duration(offsetMs) * time.Millisecond + statuses := map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected, latency: 20*time.Millisecond + offset}, + "route2": {status: peer.StatusConnected, latency: 20*time.Millisecond - offset}, + } + + chosen, _ := w.getBestRouteFromStatuses(statuses) + require.Equal(t, route.ID("route1"), chosen, "route changed on jitter at sample %d", i) + } +} diff --git a/client/internal/routemanager/client/env.go b/client/internal/routemanager/client/env.go new file mode 100644 index 000000000..398667cdf --- /dev/null +++ b/client/internal/routemanager/client/env.go @@ -0,0 +1,77 @@ +package client + +import ( + "os" + "strings" + "sync" + "time" + + log "github.com/sirupsen/logrus" +) + +const ( + // envRelayedSwitchDelay sets how long a routing peer that dropped from a + // direct to a relayed connection while carrying the route is kept before the + // route moves to an equivalent peer with a direct connection. Takes a + // duration ("30s", "0" to switch at once) or "never" to only leave such a + // peer when it disconnects. A routing peer that was already relayed when + // chosen always gives way to a direct one at once. + envRelayedSwitchDelay = "NB_HA_RELAYED_SWITCH_DELAY" + + relayedSwitchDelayNever = "never" + + // defaultRelayedSwitchDelay covers the common case of a direct connection + // that drops to relay and recovers within seconds. Moving the route + // resets every masqueraded flow using it, staying on the relay does not. + defaultRelayedSwitchDelay = 30 * time.Second +) + +// relayedSwitchPolicy is the resolved envRelayedSwitchDelay setting. +type relayedSwitchPolicy struct { + delay time.Duration + // never keeps a connected relayed routing peer regardless of delay + never bool +} + +func (p relayedSwitchPolicy) String() string { + if p.never { + return relayedSwitchDelayNever + } + return p.delay.String() +} + +var ( + relayedSwitchOnce sync.Once + resolvedRelayedSwitch relayedSwitchPolicy +) + +// relayedSwitch returns the process-wide relayed switch policy, read from the +// environment on first use. +func relayedSwitch() relayedSwitchPolicy { + relayedSwitchOnce.Do(func() { + resolvedRelayedSwitch = parseRelayedSwitchPolicy(os.Getenv(envRelayedSwitchDelay)) + log.Infof("routing peer relayed switch delay: %s", resolvedRelayedSwitch) + }) + return resolvedRelayedSwitch +} + +func parseRelayedSwitchPolicy(raw string) relayedSwitchPolicy { + def := relayedSwitchPolicy{delay: defaultRelayedSwitchDelay} + + val := strings.TrimSpace(raw) + if val == "" { + return def + } + if strings.EqualFold(val, relayedSwitchDelayNever) { + return relayedSwitchPolicy{never: true} + } + + delay, err := time.ParseDuration(val) + if err != nil || delay < 0 { + log.Warnf("invalid %s value %q, using the default of %s: want a duration or %q", + envRelayedSwitchDelay, raw, def, relayedSwitchDelayNever) + return def + } + + return relayedSwitchPolicy{delay: delay} +} diff --git a/client/internal/routemanager/client/relayed_switch_test.go b/client/internal/routemanager/client/relayed_switch_test.go new file mode 100644 index 000000000..8f4a67bba --- /dev/null +++ b/client/internal/routemanager/client/relayed_switch_test.go @@ -0,0 +1,358 @@ +package client + +import ( + "context" + "net/netip" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/netbirdio/netbird/client/internal/peer" + "github.com/netbirdio/netbird/client/proto" + "github.com/netbirdio/netbird/route" +) + +func TestParseRelayedSwitchPolicy(t *testing.T) { + testCases := []struct { + raw string + want relayedSwitchPolicy + }{ + {raw: "", want: relayedSwitchPolicy{delay: defaultRelayedSwitchDelay}}, + {raw: " ", want: relayedSwitchPolicy{delay: defaultRelayedSwitchDelay}}, + {raw: "0", want: relayedSwitchPolicy{}}, + {raw: "10s", want: relayedSwitchPolicy{delay: 10 * time.Second}}, + {raw: " 2m ", want: relayedSwitchPolicy{delay: 2 * time.Minute}}, + {raw: "never", want: relayedSwitchPolicy{never: true}}, + {raw: "NEVER", want: relayedSwitchPolicy{never: true}}, + // unrecognised values keep the default rather than guessing + {raw: "-5s", want: relayedSwitchPolicy{delay: defaultRelayedSwitchDelay}}, + {raw: "soon", want: relayedSwitchPolicy{delay: defaultRelayedSwitchDelay}}, + {raw: "30", want: relayedSwitchPolicy{delay: defaultRelayedSwitchDelay}}, + } + + for _, tc := range testCases { + t.Run(tc.raw, func(t *testing.T) { + assert.Equal(t, tc.want, parseRelayedSwitchPolicy(tc.raw), "policy parsed from %q", tc.raw) + }) + } +} + +// newRelayedTestWatcher is a watcher over two equal-metric routes, currently on +// route1, for the given policy. +func newRelayedTestWatcher(t *testing.T, policy relayedSwitchPolicy) *Watcher { + t.Helper() + + w := newTestWatcher(t, time.Time{}) + w.relayedSwitch = policy + t.Cleanup(func() { + if w.relayedHold != nil { + w.relayedHold.Stop() + } + }) + return w +} + +// degrade walks the current routing peer route1 from direct to relayed, the way +// it is seen when a direct connection carrying the route drops to relay. +func degrade(t *testing.T, w *Watcher) { + t.Helper() + + chosen, _ := w.getBestRouteFromStatuses(currentDirect) + require.Equal(t, route.ID("route1"), chosen, "route1 must carry the route while direct") + + chosen, _ = w.getBestRouteFromStatuses(currentRelayed) + require.Equal(t, route.ID("route1"), chosen, "a routing peer that just dropped to relay must be kept") + require.False(t, w.degradedSince.IsZero(), "the drop to relay must be recorded") +} + +var ( + // route1 is relayed, route2 is an equivalent peer with a direct connection + currentRelayed = map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected, relayed: true, latency: 50 * time.Millisecond}, + "route2": {status: peer.StatusConnected, latency: 60 * time.Millisecond}, + } + currentDirect = map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected, latency: 50 * time.Millisecond}, + "route2": {status: peer.StatusConnected, latency: 60 * time.Millisecond}, + } +) + +func TestDegradedRoutingPeerIsKeptDuringDelay(t *testing.T) { + w := newRelayedTestWatcher(t, relayedSwitchPolicy{delay: 30 * time.Second}) + + degrade(t, w) + require.NotNil(t, w.relayedHold, "the pending switch must arm a timer so it is applied when due") + + // the direct connection comes back within the delay: nothing moved, and the + // degraded period ends + chosen, _ := w.getBestRouteFromStatuses(currentDirect) + assert.Equal(t, route.ID("route1"), chosen, "the recovered routing peer must stay chosen") + assert.True(t, w.degradedSince.IsZero(), "the degraded period must end once the peer is direct again") +} + +func TestDegradedRoutingPeerIsReplacedAfterDelay(t *testing.T) { + w := newRelayedTestWatcher(t, relayedSwitchPolicy{delay: 30 * time.Second}) + + degrade(t, w) + w.degradedSince = w.degradedSince.Add(-31 * time.Second) + + chosen, _ := w.getBestRouteFromStatuses(currentRelayed) + assert.Equal(t, route.ID("route2"), chosen, "a routing peer relayed for longer than the delay must give way to a direct one") + assert.Equal(t, switchReasonDirect, w.switchReason, "the switch reason must name the relayed connection") +} + +// A routing peer that was relayed when it was chosen is not degraded: when an +// equivalent direct peer shows up the route must upgrade at once instead of +// staying on the slower path for the delay. +func TestRelayedRoutingPeerUpgradesAtOnce(t *testing.T) { + w := newRelayedTestWatcher(t, relayedSwitchPolicy{never: true}) + + // route1 was chosen while relayed and route2 was still connecting + chosen, _ := w.getBestRouteFromStatuses(map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected, relayed: true}, + "route2": {status: peer.StatusConnecting}, + }) + require.Equal(t, route.ID("route1"), chosen) + require.True(t, w.degradedSince.IsZero(), "a peer relayed from the start is not degraded") + + chosen, _ = w.getBestRouteFromStatuses(currentRelayed) + assert.Equal(t, route.ID("route2"), chosen, "a direct peer must take over from one that was relayed all along") + assert.Equal(t, switchReasonDirect, w.switchReason) + assert.Nil(t, w.relayedHold, "an upgrade must not schedule anything") +} + +// An idle routing peer woken by traffic whose relay connection comes up before +// the direct one was never direct: it is not degraded, and must give way to an +// equivalent direct peer at once. +func TestIdleRoutingPeerWokenOverRelayUpgradesAtOnce(t *testing.T) { + w := newRelayedTestWatcher(t, relayedSwitchPolicy{never: true}) + + chosen, _ := w.getBestRouteFromStatuses(map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusIdle}, + "route2": {status: peer.StatusIdle}, + }) + require.Equal(t, route.ID("route1"), chosen) + + chosen, _ = w.getBestRouteFromStatuses(map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected, relayed: true}, + "route2": {status: peer.StatusConnecting}, + }) + require.Equal(t, route.ID("route1"), chosen) + require.True(t, w.degradedSince.IsZero(), "waking up over relay is not a drop from direct") + + chosen, _ = w.getBestRouteFromStatuses(currentRelayed) + assert.Equal(t, route.ID("route2"), chosen, "a direct peer must take over from one that was never direct") + assert.Nil(t, w.relayedHold, "an upgrade must not schedule anything") +} + +// A routing peer that drops to relay again, after recovering from an earlier +// drop, is a new degradation and must be reported at info level again. +func TestRepeatedDegradationIsReportedAgain(t *testing.T) { + w := newRelayedTestWatcher(t, relayedSwitchPolicy{delay: 30 * time.Second}) + + degrade(t, w) + require.Equal(t, route.ID("route2"), w.heldCandidate, "the first drop must be reported") + + // the first report is well within the report interval + w.heldReported = w.heldReported.Add(-time.Minute) + firstReport := w.heldReported + + degrade(t, w) + assert.True(t, w.heldReported.After(firstReport), "a new drop to relay must be reported, not treated as a repeat") +} + +func TestDegradedRoutingPeerNeverPolicy(t *testing.T) { + w := newRelayedTestWatcher(t, relayedSwitchPolicy{never: true}) + + degrade(t, w) + w.degradedSince = w.degradedSince.Add(-time.Hour) + + chosen, _ := w.getBestRouteFromStatuses(currentRelayed) + assert.Equal(t, route.ID("route1"), chosen, "with never, a degraded routing peer is kept while connected") + assert.Nil(t, w.relayedHold, "with never there is nothing to schedule") +} + +func TestDegradedRoutingPeerZeroDelaySwitchesAtOnce(t *testing.T) { + w := newRelayedTestWatcher(t, relayedSwitchPolicy{}) + + chosen, _ := w.getBestRouteFromStatuses(currentDirect) + require.Equal(t, route.ID("route1"), chosen) + + chosen, _ = w.getBestRouteFromStatuses(currentRelayed) + assert.Equal(t, route.ID("route2"), chosen, "a zero delay must keep the immediate switch to a direct peer") +} + +// The delay only covers a candidate that is better solely by being direct. A +// better metric or a lost routing peer must move the route at once. +func TestRelayedDelayDoesNotHoldOtherSwitches(t *testing.T) { + t.Run("better metric", func(t *testing.T) { + w := newRelayedTestWatcher(t, relayedSwitchPolicy{never: true}) + degrade(t, w) + w.routes["route2"].Metric = route.MaxMetric - 1 + + chosen, _ := w.getBestRouteFromStatuses(currentRelayed) + assert.Equal(t, route.ID("route2"), chosen, "a better metric must win over a degraded peer immediately") + assert.Equal(t, switchReasonMetric, w.switchReason) + }) + + t.Run("current peer disconnected", func(t *testing.T) { + w := newRelayedTestWatcher(t, relayedSwitchPolicy{never: true}) + degrade(t, w) + + chosen, _ := w.getBestRouteFromStatuses(map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnecting, relayed: true}, + "route2": {status: peer.StatusConnected, latency: 60 * time.Millisecond}, + }) + assert.Equal(t, route.ID("route2"), chosen, "a routing peer that is gone must be replaced regardless of the delay") + assert.Equal(t, switchReasonUnavailable, w.switchReason) + }) + + t.Run("both relayed", func(t *testing.T) { + w := newRelayedTestWatcher(t, relayedSwitchPolicy{delay: 30 * time.Second}) + + chosen, _ := w.getBestRouteFromStatuses(map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected, relayed: true}, + "route2": {status: peer.StatusConnected, relayed: true}, + }) + assert.Equal(t, route.ID("route1"), chosen, "equally relayed peers are a tie and the current one stays") + assert.Nil(t, w.relayedHold, "a tie must not schedule a switch") + }) +} + +// TestRelayedHoldTimerAppliesSwitch drives the Start loop: once the delay has +// passed, the held switch must be applied without waiting for a peer update or +// the periodic re-evaluation. +func TestRelayedHoldTimerAppliesSwitch(t *testing.T) { + w, handler := newRecalcWatcher(t) + w.relayedSwitch = relayedSwitchPolicy{delay: 50 * time.Millisecond} + + recorder := w.statusRecorder + for _, key := range []string{"peer1", "peer2"} { + require.NoError(t, recorder.AddPeer(key, key+".netbird.cloud", "100.64.0.1", "")) + } + require.NoError(t, recorder.UpdatePeerState(peer.State{PubKey: "peer1", ConnStatus: peer.StatusConnected, Relayed: true})) + require.NoError(t, recorder.UpdatePeerState(peer.State{PubKey: "peer2", ConnStatus: peer.StatusConnected})) + + require.NoError(t, w.recalculateRoutes(reasonPeerUpdate, map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected}, + })) + require.Equal(t, []string{"peer1"}, handler.added) + + // route1 drops to relay while route2 is direct: held for the delay + require.NoError(t, w.recalculateRoutes(reasonPeerUpdate, map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected, relayed: true}, + "route2": {status: peer.StatusConnected}, + })) + require.Equal(t, []string{"peer1"}, handler.added, "the switch must be held while the delay runs") + + w.ctx, w.cancel = context.WithCancel(context.Background()) + w.done = make(chan struct{}) + go w.Start() + t.Cleanup(w.Stop) + + // the loop owns the watcher state now, so observe the switch through the + // event it publishes rather than reading its fields + require.Eventually(t, func() bool { + for _, e := range recorder.GetEventHistory() { + if e.Message == "Routing peer changed" && e.Metadata["peer"] == "peer2" { + return true + } + } + return false + }, 2*time.Second, 10*time.Millisecond, "the held switch must be applied once the delay passed") +} + +// A route update can keep a route's ID and hand it to a different routing peer. +// That is a switch like any other: it is reported, and the new peer is tracked +// from its own state rather than inheriting the previous peer's. +func TestRouteReassignedToDifferentPeerIsASwitch(t *testing.T) { + w, handler := newRecalcWatcher(t) + + require.NoError(t, w.recalculateRoutes(reasonPeerUpdate, map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected}, + })) + require.Equal(t, []string{"peer1"}, handler.added) + + reassigned := *w.routes["route1"] + reassigned.Peer = "peer3" + w.routes["route1"] = &reassigned + + require.NoError(t, w.recalculateRoutes(reasonRouteUpdate, map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected, relayed: true}, + })) + assert.Equal(t, []string{"peer1", "peer3"}, handler.added, "the allowed IPs must move to the new routing peer") + assert.Equal(t, "peer3", w.trackedPeer, "the new routing peer must be tracked") + assert.False(t, w.trackedDirect, "the new routing peer is tracked from its own state") + + events := w.statusRecorder.GetEventHistory() + require.Len(t, events, 1, "the reassignment must publish a switch event") + assert.Equal(t, "Routing peer changed", events[0].Message) + assert.Equal(t, "peer1", events[0].Metadata["previous_peer"]) + assert.Equal(t, "peer3", events[0].Metadata["peer"]) + assert.Equal(t, string(switchReasonRouteChanged), events[0].Metadata["reason"]) +} + +// Exit nodes report their own connect and disconnect events, but those carry +// no reason: the switch event is published for default routes too. +func TestDefaultRouteSwitchPublishesEvent(t *testing.T) { + w, _ := newRecalcWatcher(t) + for _, r := range w.routes { + r.Network = netip.MustParsePrefix("0.0.0.0/0") + } + + require.NoError(t, w.recalculateRoutes(reasonPeerUpdate, map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected}, + })) + require.NoError(t, w.recalculateRoutes(reasonPeerUpdate, map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnecting}, + "route2": {status: peer.StatusConnected}, + })) + + var switchEvents []*proto.SystemEvent + for _, e := range w.statusRecorder.GetEventHistory() { + if e.Message == "Routing peer changed" { + switchEvents = append(switchEvents, e) + } + } + require.Len(t, switchEvents, 1, "an exit node switch must publish a switch event") + assert.Equal(t, string(switchReasonUnavailable), switchEvents[0].Metadata["reason"]) + assert.Empty(t, switchEvents[0].UserMessage, "the switch event must not notify the user a second time") +} + +func TestRoutingPeerSwitchPublishesEvent(t *testing.T) { + w, _ := newRecalcWatcher(t) + + require.NoError(t, w.recalculateRoutes(reasonPeerUpdate, map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected, latency: 20 * time.Millisecond}, + })) + assert.Empty(t, w.statusRecorder.GetEventHistory(), "the initial assignment is not a switch") + + require.NoError(t, w.recalculateRoutes(reasonPeerUpdate, map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnecting}, + "route2": {status: peer.StatusConnected, latency: 20 * time.Millisecond}, + })) + + events := w.statusRecorder.GetEventHistory() + require.Len(t, events, 1, "a switch must publish exactly one event") + assert.Equal(t, proto.SystemEvent_WARNING, events[0].Severity) + assert.Equal(t, "Routing peer changed", events[0].Message) + assert.Equal(t, map[string]string{ + "network": "recording", + "id": "", + "previous_peer": "peer1", + "peer": "peer2", + "reason": string(switchReasonUnavailable), + }, events[0].Metadata, "the event must name both peers and the reason") + + require.NoError(t, w.recalculateRoutes(reasonPeerUpdate, map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnecting}, + "route2": {status: peer.StatusConnecting}, + })) + + events = w.statusRecorder.GetEventHistory() + require.Len(t, events, 2, "losing the last routing peer must publish an event") + assert.Equal(t, "No routing peer available", events[1].Message) + assert.Equal(t, "peer2", events[1].Metadata["previous_peer"]) +}