diff --git a/client/internal/routemanager/client/client.go b/client/internal/routemanager/client/client.go index b882c4d4a..66c2ed491 100644 --- a/client/internal/routemanager/client/client.go +++ b/client/internal/routemanager/client/client.go @@ -65,6 +65,12 @@ const ( // well above any usable path so that slow paths keep their order. unknownLatency = 10 * time.Second + // unmeasuredLatencyHold is how long a current routing peer without a + // latency sample is kept against a measured one. A direct connection is + // sampled within seconds, so a peer still unsampled after this is ranked + // at unknownLatency rather than held indefinitely. + unmeasuredLatencyHold = 30 * 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 @@ -99,6 +105,7 @@ type holdKind int const ( holdLatencyMargin holdKind = iota holdLatencyDwell + holdLatencyUnmeasured holdRelayedDelay ) @@ -124,6 +131,9 @@ type routeCandidate struct { relayed bool latency time.Duration noise time.Duration + // measured is false while the peer has no latency sample, as right after a + // direct connection comes up: latency then holds unknownLatency + measured bool } func newRouteCandidate(r *route.Route, status routerPeerStatus) routeCandidate { @@ -140,6 +150,7 @@ func newRouteCandidate(r *route.Route, status routerPeerStatus) routeCandidate { relayed: status.relayed, latency: latency, noise: status.noise, + measured: status.latency > 0, } } @@ -230,6 +241,9 @@ type WatcherConfig struct { StatusRecorder *peer.Status Route *route.Route Handler RouteHandler + // ReconcileAllowedIPs re-applies the routed allowed IPs tracked for a + // routing peer onto the interface. Optional. + ReconcileAllowedIPs func(peerKey string) error } // Watcher watches route and peer changes and updates allowed IPs accordingly. @@ -271,11 +285,20 @@ type Watcher struct { 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 is when the relayedSwitch delay of the current relayed + // routing peer started: its drop from direct, or the first direct contender + // if it was never direct. Zero while it is direct or uncontended. 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 + // unmeasuredRoute and unmeasuredSince track since when the current route's + // peer has had no latency sample, bounding the hold in shouldSwitchRoute + unmeasuredRoute route.ID + unmeasuredSince time.Time + // neverDirect marks a degradedSince started by a direct contender rather + // than by a drop from direct, which the "never" policy does not cover + neverDirect bool + // relayedHold fires when a switch held back by the relayed delay or the + // unmeasured latency hold 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 @@ -298,6 +321,9 @@ type Watcher struct { stopped bool handler RouteHandler updateSerial uint64 + // reconcileAllowedIPs re-applies a routing peer's routed allowed IPs when + // it connects while keeping the route, see keepChosenRoute. May be nil. + reconcileAllowedIPs func(peerKey string) error } func newPeerStateUpdate() chan struct{} { @@ -320,6 +346,7 @@ func NewWatcher(config WatcherConfig) *Watcher { handler: config.Handler, currentChosenStatus: nil, relayedSwitch: relayedSwitch(), + reconcileAllowedIPs: config.ReconcileAllowedIPs, } return client } @@ -393,6 +420,8 @@ func (w *Watcher) getBestRouteFromStatuses(routePeerStatuses map[route.ID]router } w.trackRelayed(current, haveCurrent) + w.trackUnmeasured(current, haveCurrent) + w.trackContender(current, best, haveCurrent && haveBest) if !haveBest { w.reportUnassigned() @@ -449,9 +478,8 @@ func (w *Watcher) reportUnassigned() { // 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. +// delay at once. A peer that was never direct starts it later, when a direct +// peer first contends for the route, see relayedSwitchDue. func (w *Watcher) trackRelayed(current routeCandidate, haveCurrent bool) { if !haveCurrent { w.trackedCurrent, w.trackedPeer, w.trackedDirect, w.degradedSince = "", "", false, time.Time{} @@ -462,7 +490,7 @@ func (w *Watcher) trackRelayed(current routeCandidate, haveCurrent bool) { case current.id != w.trackedCurrent, current.peer != w.trackedPeer, !current.relayed: w.degradedSince = time.Time{} case w.trackedDirect: - w.degradedSince = time.Now() + w.degradedSince, w.neverDirect = time.Now(), false // a new degradation, reported as such even if an earlier one held // back the same candidate w.heldCandidate = "" @@ -494,6 +522,20 @@ func (w *Watcher) shouldSwitchRoute(current, candidate routeCandidate) bool { return true } + // Without a sample on both sides there is no gain to weigh. A direct + // connection reports no latency until its first sample, so comparing it as + // unknownLatency would move the route away from a peer that just recovered. + // The hold is bounded so a peer that never gets a sample does not keep the + // route for good. + if w.unmeasuredHoldActive(current, candidate) { + if level := w.holdLevel(candidate, holdLatencyUnmeasured); log.IsLevelEnabled(level) { + log.StandardLogger().Logf(level, + "keeping routing peer %s for [%v] over %s: latency not measured on both peers yet (%s)", + current.peer, w.handler, candidate.peer, candidate) + } + return false + } + gain := current.latency - candidate.latency if margin := switchMargin(current, candidate); gain < margin { if level := w.holdLevel(candidate, holdLatencyMargin); log.IsLevelEnabled(level) { @@ -518,22 +560,75 @@ func (w *Watcher) shouldSwitchRoute(current, candidate routeCandidate) bool { return true } +// trackContender restarts the relayed delay of a peer that was never direct +// once no direct contender is left. That delay runs from the first contender, +// so keeping it would let a later contender take over without waiting. The +// delay of a peer that dropped from direct keeps running from the drop. +func (w *Watcher) trackContender(current, best routeCandidate, have bool) { + if !w.neverDirect || w.degradedSince.IsZero() { + return + } + if have && best.winsOnlyOnRelay(current) { + return + } + w.degradedSince = time.Time{} +} + +// trackUnmeasured records since when the current route's peer has had no +// latency sample, and forgets it once the peer is measured or the route moves. +func (w *Watcher) trackUnmeasured(current routeCandidate, haveCurrent bool) { + if !haveCurrent || current.measured || current.id != w.unmeasuredRoute { + w.unmeasuredRoute, w.unmeasuredSince = "", time.Time{} + } + if haveCurrent && !current.measured && w.unmeasuredSince.IsZero() { + w.unmeasuredRoute, w.unmeasuredSince = current.id, time.Now() + } +} + +// unmeasuredHoldActive reports whether a latency comparison must wait because +// a side has no sample: always for an unsampled candidate, and for an unsampled +// current peer until unmeasuredLatencyHold runs out. A pending hold arms +// relayedHold so the switch is reconsidered when it runs out. +func (w *Watcher) unmeasuredHoldActive(current, candidate routeCandidate) bool { + if !candidate.measured { + return true + } + if current.measured { + return false + } + + remaining := unmeasuredLatencyHold - time.Since(w.unmeasuredSince) + if remaining <= 0 { + return false + } + w.armRelayedHold(remaining) + + 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. +// A relayed peer is kept for the relayedSwitch delay: the relay still carries +// the traffic, and staying keeps every flow through this routing peer alive. +// For a peer that dropped from direct to relayed the delay runs from the drop, +// since a direct connection that dropped often recovers within seconds. For a +// peer that was never direct, as when the peers of a group come up over relay +// before their direct connections, it runs from the first evaluation that +// finds a direct contender: the current peer usually turns direct within +// seconds too. 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 + w.degradedSince, w.neverDirect = time.Now(), true } policy := w.relayedSwitch + // "never" keeps a peer whose direct connection dropped; one that was never + // direct still moves after the default delay instead of staying relayed + if policy.never && w.neverDirect { + policy = relayedSwitchPolicy{delay: defaultRelayedSwitchDelay} + } if !policy.never { remaining := policy.delay - time.Since(w.degradedSince) if remaining <= 0 { @@ -544,7 +639,7 @@ func (w *Watcher) relayedSwitchDue(current, candidate routeCandidate) bool { 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", + "keeping relayed routing peer %s for [%v] over direct %s: held for %v, switch delay %s", current.peer, w.handler, candidate.peer, time.Since(w.degradedSince).Round(time.Second), policy) } @@ -677,21 +772,46 @@ func (w *Watcher) removeAllowedIPs(route *route.Route, rsn reason) error { return nil } -// shouldSkipRecalculation checks if we can skip route recalculation for the same route without status changes -func (w *Watcher) shouldSkipRecalculation(newChosenID route.ID, newStatus routerPeerStatus) bool { - if w.currentChosen == nil { +// keepChosenRoute reports whether the evaluation keeps the current route, with +// the same definition, on the same routing peer. A status change of that peer +// is recorded without removing and re-adding its allowed IPs, which would leave +// the prefix without an owner in between. When the peer connects, its routed +// allowed IPs are re-applied instead: a peer recreated by a peer update comes +// back with its overlay addresses only. +func (w *Watcher) keepChosenRoute(newChosenID route.ID, newStatus routerPeerStatus) bool { + if w.currentChosen == nil || w.currentChosen.ID != newChosenID { return false } - isSameRoute := w.currentChosen.ID == newChosenID && w.currentChosen.Equal(w.routes[newChosenID]) - if !isSameRoute { + newChosenRoute := w.routes[newChosenID] + if !w.currentChosen.Equal(newChosenRoute) { return false } - if w.currentChosenStatus != nil { - return w.currentChosenStatus.status == newStatus.status + if w.currentChosenStatus == nil || w.currentChosenStatus.status == newStatus.status { + return true } + // a failed reconcile leaves the previous status recorded, so the next + // evaluation sees the change again and retries + if newStatus.status == peer.StatusConnected && !w.reconcileRoutedAllowedIPs(newChosenRoute.Peer) { + return true + } + + w.currentChosen = newChosenRoute + w.currentChosenStatus = &newStatus + + return true +} + +func (w *Watcher) reconcileRoutedAllowedIPs(peerKey string) bool { + if w.reconcileAllowedIPs == nil { + return true + } + if err := w.reconcileAllowedIPs(peerKey); err != nil { + log.Warnf("failed to reconcile routed allowed IPs of routing peer %s for [%v], will retry: %v", peerKey, w.handler, err) + return false + } return true } @@ -719,8 +839,7 @@ func (w *Watcher) recalculateRoutes(rsn reason, routerPeerStatuses map[route.ID] return nil } - // If we can skip recalculation for the same route without changes, do nothing - if w.shouldSkipRecalculation(newChosenID, newStatus) { + if w.keepChosenRoute(newChosenID, newStatus) { return nil } diff --git a/client/internal/routemanager/client/client_test.go b/client/internal/routemanager/client/client_test.go index 764a2f74d..641383e90 100644 --- a/client/internal/routemanager/client/client_test.go +++ b/client/internal/routemanager/client/client_test.go @@ -2,6 +2,7 @@ package client import ( "context" + "errors" "fmt" "net/netip" "sync" @@ -1480,3 +1481,84 @@ func TestLatencyJitterDoesNotSwitchRoute(t *testing.T) { require.Equal(t, route.ID("route1"), chosen, "route changed on jitter at sample %d", i) } } + +// A routing peer that keeps the route while only its connection status changes, +// as when an idle peer is woken by traffic, must keep its allowed IPs in place: +// removing and re-adding them leaves the prefix without an owner in between. +func TestRecalculateRoutesStatusChangeKeepsAllowedIPs(t *testing.T) { + w, handler := newRecalcWatcher(t) + + require.NoError(t, w.recalculateRoutes(reasonPeerUpdate, map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusIdle}, + })) + require.Equal(t, []string{"peer1"}, handler.added) + + require.NoError(t, w.recalculateRoutes(reasonPeerUpdate, map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected, relayed: true}, + })) + assert.Equal(t, []string{"peer1"}, handler.added, "a status change must not re-add the allowed IPs") + assert.Equal(t, 0, handler.removed, "a status change must not remove the allowed IPs") + require.NotNil(t, w.currentChosenStatus) + assert.Equal(t, peer.StatusConnected, w.currentChosenStatus.status, "the new status must be recorded") +} + +// A routing peer recreated by a peer update comes back idle with its overlay +// addresses only. When it connects while keeping the route, its routed allowed +// IPs must be re-applied, since nothing else puts them back on the new peer. +func TestRecalculateRoutesReconcilesAllowedIPsOnConnect(t *testing.T) { + w, handler := newRecalcWatcher(t) + var reconciled []string + w.reconcileAllowedIPs = func(peerKey string) error { + reconciled = append(reconciled, peerKey) + return nil + } + + require.NoError(t, w.recalculateRoutes(reasonPeerUpdate, map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected}, + })) + require.Equal(t, []string{"peer1"}, handler.added) + require.Empty(t, reconciled, "a fresh assignment adds the allowed IPs itself") + + require.NoError(t, w.recalculateRoutes(reasonPeerUpdate, map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusIdle}, + })) + assert.Empty(t, reconciled, "going idle must not reconcile") + + require.NoError(t, w.recalculateRoutes(reasonPeerUpdate, map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected}, + })) + assert.Equal(t, []string{"peer1"}, reconciled, "connecting while keeping the route must reconcile its allowed IPs") + assert.Equal(t, []string{"peer1"}, handler.added, "the reconcile must not go through the refcounted add") + assert.Equal(t, 0, handler.removed, "the allowed IPs must not be removed") +} + +// A failed reconcile must be retried: the new status is only recorded once the +// routed allowed IPs are back, so the next evaluation tries again. +func TestRecalculateRoutesRetriesFailedReconcile(t *testing.T) { + w, _ := newRecalcWatcher(t) + calls := 0 + w.reconcileAllowedIPs = func(string) error { + calls++ + if calls == 1 { + return errors.New("add allowed IP: transient failure") + } + return nil + } + + connected := map[route.ID]routerPeerStatus{"route1": {status: peer.StatusConnected}} + require.NoError(t, w.recalculateRoutes(reasonPeerUpdate, connected)) + require.NoError(t, w.recalculateRoutes(reasonPeerUpdate, map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusIdle}, + })) + + require.NoError(t, w.recalculateRoutes(reasonPeerUpdate, connected)) + require.Equal(t, 1, calls, "connecting must reconcile") + assert.Equal(t, peer.StatusIdle, w.currentChosenStatus.status, "a failed reconcile must not record the new status") + + require.NoError(t, w.recalculateRoutes(reasonPeerUpdate, connected)) + assert.Equal(t, 2, calls, "the next evaluation must retry the reconcile") + assert.Equal(t, peer.StatusConnected, w.currentChosenStatus.status, "a successful reconcile records the status") + + require.NoError(t, w.recalculateRoutes(reasonPeerUpdate, connected)) + assert.Equal(t, 2, calls, "nothing to retry once the reconcile succeeded") +} diff --git a/client/internal/routemanager/client/env.go b/client/internal/routemanager/client/env.go index 398667cdf..6a7c08a1f 100644 --- a/client/internal/routemanager/client/env.go +++ b/client/internal/routemanager/client/env.go @@ -12,10 +12,11 @@ import ( 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. + // route moves to an equivalent peer with a direct connection. A routing peer + // that was never direct is kept for the same delay from when a direct peer + // first contends for the route. Takes a duration ("30s", "0" to switch at + // once) or "never" to only leave a peer that dropped from direct when it + // disconnects; a peer that was never direct then uses the default delay. envRelayedSwitchDelay = "NB_HA_RELAYED_SWITCH_DELAY" relayedSwitchDelayNever = "never" diff --git a/client/internal/routemanager/client/relayed_switch_test.go b/client/internal/routemanager/client/relayed_switch_test.go index 8f4a67bba..f70dfff50 100644 --- a/client/internal/routemanager/client/relayed_switch_test.go +++ b/client/internal/routemanager/client/relayed_switch_test.go @@ -103,11 +103,11 @@ func TestDegradedRoutingPeerIsReplacedAfterDelay(t *testing.T) { 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}) +// A routing peer that was relayed when it was chosen is held for the delay when +// an equivalent direct peer shows up, from that moment on: it usually turns +// direct within seconds, and moving the route resets its masqueraded flows. +func TestRelayedRoutingPeerIsKeptDuringDelay(t *testing.T) { + w := newRelayedTestWatcher(t, relayedSwitchPolicy{delay: 30 * time.Second}) // route1 was chosen while relayed and route2 was still connecting chosen, _ := w.getBestRouteFromStatuses(map[route.ID]routerPeerStatus{ @@ -115,36 +115,57 @@ func TestRelayedRoutingPeerUpgradesAtOnce(t *testing.T) { "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") + require.True(t, w.degradedSince.IsZero(), "the delay must not run before a direct peer contends") 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") + assert.Equal(t, route.ID("route1"), chosen, "a relayed peer must be kept when a direct one shows up") + assert.False(t, w.degradedSince.IsZero(), "the direct contender must start the delay") + assert.NotNil(t, w.relayedHold, "the pending switch must arm a timer so it is applied when due") + + w.degradedSince = w.degradedSince.Add(-31 * time.Second) + chosen, _ = w.getBestRouteFromStatuses(currentRelayed) + assert.Equal(t, route.ID("route2"), chosen, "a peer still relayed after the delay must give way to a direct one") + assert.Equal(t, switchReasonDirect, w.switchReason, "the switch reason must name the relayed connection") } -// 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}) +// The peers of an HA group woken from idle come up over relay first and turn +// direct one after the other. The route must stay on the peer it was on when +// the other one turns direct first and the current one follows within the +// delay, including before the current peer's first latency sample. +func TestIdleRoutingPeersWokenOverRelayKeepRoute(t *testing.T) { + w := newRelayedTestWatcher(t, relayedSwitchPolicy{delay: 30 * time.Second}) - chosen, _ := w.getBestRouteFromStatuses(map[route.ID]routerPeerStatus{ - "route1": {status: peer.StatusIdle}, - "route2": {status: peer.StatusIdle}, - }) - require.Equal(t, route.ID("route1"), chosen) + steps := []struct { + name string + statuses map[route.ID]routerPeerStatus + }{ + {"both idle", map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusIdle}, + "route2": {status: peer.StatusIdle}, + }}, + {"both relayed", map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected, relayed: true}, + "route2": {status: peer.StatusConnected, relayed: true}, + }}, + {"other direct first", map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected, relayed: true}, + "route2": {status: peer.StatusConnected, latency: 20 * time.Millisecond}, + }}, + {"current direct, not sampled", map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected}, + "route2": {status: peer.StatusConnected, latency: 20 * time.Millisecond}, + }}, + {"current direct, sampled", map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected, latency: 38 * time.Millisecond}, + "route2": {status: peer.StatusConnected, latency: 20 * time.Millisecond}, + }}, + } - 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") + for _, step := range steps { + chosen, _ := w.getBestRouteFromStatuses(step.statuses) + require.Equal(t, route.ID("route1"), chosen, "step %q must keep the route on route1", step.name) + } + assert.True(t, w.degradedSince.IsZero(), "the delay must end once the current peer is direct") } // A routing peer that drops to relay again, after recovering from an earlier @@ -356,3 +377,128 @@ func TestRoutingPeerSwitchPublishesEvent(t *testing.T) { assert.Equal(t, "No routing peer available", events[1].Message) assert.Equal(t, "peer2", events[1].Metadata["previous_peer"]) } + +// The statuses below follow the status recorder: a drop to relay clears the +// latency, and a recovered direct connection reports no latency until the new +// ICE agent's first sample lands. +var ( + flapDirect = map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected, latency: 50 * time.Millisecond}, + "route2": {status: peer.StatusConnected, latency: 60 * time.Millisecond}, + } + flapRelayed = map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected, relayed: true}, + "route2": {status: peer.StatusConnected, latency: 60 * time.Millisecond}, + } + flapRecoveredUnsampled = map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected}, + "route2": {status: peer.StatusConnected, latency: 60 * time.Millisecond}, + } +) + +// A routing peer whose direct connection drops to relay and comes back within +// the delay must keep the route, including the window between the recovery and +// the first latency sample of the new connection. +func TestRecoveredRoutingPeerWithoutLatencySampleIsKept(t *testing.T) { + w := newRelayedTestWatcher(t, relayedSwitchPolicy{delay: 30 * time.Second}) + + chosen, _ := w.getBestRouteFromStatuses(flapDirect) + require.Equal(t, route.ID("route1"), chosen) + + chosen, _ = w.getBestRouteFromStatuses(flapRelayed) + require.Equal(t, route.ID("route1"), chosen, "a routing peer that just dropped to relay must be kept") + + chosen, _ = w.getBestRouteFromStatuses(flapRecoveredUnsampled) + assert.Equal(t, route.ID("route1"), chosen, "a recovered routing peer must not lose the route before its latency is sampled") +} + +// A routing peer that keeps dropping to relay for less than the delay, and +// staying direct for less than the delay in between, must keep the route on +// every cycle: each drop is a new degradation with its own delay. +func TestFlappingRoutingPeerIsKept(t *testing.T) { + w := newRelayedTestWatcher(t, relayedSwitchPolicy{delay: 30 * time.Second}) + + chosen, _ := w.getBestRouteFromStatuses(flapDirect) + require.Equal(t, route.ID("route1"), chosen) + + for cycle := 0; cycle < 3; cycle++ { + chosen, _ = w.getBestRouteFromStatuses(flapRelayed) + require.Equal(t, route.ID("route1"), chosen, "cycle %d: drop to relay", cycle) + + // relayed for 25s, still within the delay + w.degradedSince = w.degradedSince.Add(-25 * time.Second) + chosen, _ = w.getBestRouteFromStatuses(flapRelayed) + require.Equal(t, route.ID("route1"), chosen, "cycle %d: relayed within the delay", cycle) + + chosen, _ = w.getBestRouteFromStatuses(flapRecoveredUnsampled) + require.Equal(t, route.ID("route1"), chosen, "cycle %d: recovered, latency not sampled yet", cycle) + + chosen, _ = w.getBestRouteFromStatuses(flapDirect) + require.Equal(t, route.ID("route1"), chosen, "cycle %d: recovered and sampled", cycle) + } +} + +// "never" only keeps a routing peer that dropped from direct. A peer that was +// never direct must still move to a direct one after the default delay, so the +// route is not left on a relay for good. +func TestNeverPolicyDoesNotPinNeverDirectPeer(t *testing.T) { + w := newRelayedTestWatcher(t, relayedSwitchPolicy{never: true}) + + chosen, _ := w.getBestRouteFromStatuses(map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected, relayed: true}, + "route2": {status: peer.StatusConnecting}, + }) + require.Equal(t, route.ID("route1"), chosen) + + chosen, _ = w.getBestRouteFromStatuses(currentRelayed) + require.Equal(t, route.ID("route1"), chosen, "a never-direct peer must be held for the default delay") + require.NotNil(t, w.relayedHold, "the default delay must arm the timer") + + w.degradedSince = w.degradedSince.Add(-defaultRelayedSwitchDelay - time.Second) + chosen, _ = w.getBestRouteFromStatuses(currentRelayed) + assert.Equal(t, route.ID("route2"), chosen, "with never, a never-direct peer must still give way after the default delay") +} + +// A current peer that never gets a latency sample must not keep the route +// forever against a measured peer: once the hold runs out it is ranked as +// unknown latency again. +func TestUnmeasuredRoutingPeerHoldIsBounded(t *testing.T) { + w := newRelayedTestWatcher(t, relayedSwitchPolicy{delay: 30 * time.Second}) + + chosen, _ := w.getBestRouteFromStatuses(flapRecoveredUnsampled) + require.Equal(t, route.ID("route1"), chosen, "an unsampled current peer must be held at first") + require.False(t, w.unmeasuredSince.IsZero(), "the hold must record when it started") + + w.unmeasuredSince = w.unmeasuredSince.Add(-unmeasuredLatencyHold - time.Second) + chosen, _ = w.getBestRouteFromStatuses(flapRecoveredUnsampled) + assert.Equal(t, route.ID("route2"), chosen, "a peer still unsampled after the hold must give way to a measured one") + assert.Equal(t, switchReasonLatency, w.switchReason) +} + +// The delay of a peer that was never direct runs from the first direct +// contender. When that contender goes away the delay must start over, so a +// later contender is held for the full delay instead of taking over at once. +func TestNeverDirectDelayRestartsWhenContenderGoes(t *testing.T) { + w := newRelayedTestWatcher(t, relayedSwitchPolicy{delay: 30 * time.Second}) + + relayedAlone := map[route.ID]routerPeerStatus{ + "route1": {status: peer.StatusConnected, relayed: true}, + "route2": {status: peer.StatusConnecting}, + } + chosen, _ := w.getBestRouteFromStatuses(relayedAlone) + require.Equal(t, route.ID("route1"), chosen) + + chosen, _ = w.getBestRouteFromStatuses(currentRelayed) + require.Equal(t, route.ID("route1"), chosen, "the first direct contender must be held") + require.False(t, w.degradedSince.IsZero()) + + // the first contender was seen longer ago than the delay + w.degradedSince = w.degradedSince.Add(-31 * time.Second) + + chosen, _ = w.getBestRouteFromStatuses(relayedAlone) + require.Equal(t, route.ID("route1"), chosen) + assert.True(t, w.degradedSince.IsZero(), "the delay must start over once no direct contender is left") + + chosen, _ = w.getBestRouteFromStatuses(currentRelayed) + assert.Equal(t, route.ID("route1"), chosen, "a new direct contender must be held for the full delay") +} diff --git a/client/internal/routemanager/manager.go b/client/internal/routemanager/manager.go index 981b0c987..7d699fbb4 100644 --- a/client/internal/routemanager/manager.go +++ b/client/internal/routemanager/manager.go @@ -578,12 +578,13 @@ func (m *DefaultManager) TriggerSelection(networks route.HAMap) { } config := client.WatcherConfig{ - Context: m.ctx, - DNSRouteInterval: m.dnsRouteInterval, - WGInterface: m.wgInterface, - StatusRecorder: m.statusRecorder, - Route: routes[0], - Handler: handler, + Context: m.ctx, + DNSRouteInterval: m.dnsRouteInterval, + WGInterface: m.wgInterface, + StatusRecorder: m.statusRecorder, + Route: routes[0], + Handler: handler, + ReconcileAllowedIPs: m.ReconcilePeerAllowedIPs, } clientNetworkWatcher := client.NewWatcher(config) m.clientNetworks[id] = clientNetworkWatcher @@ -628,12 +629,13 @@ func (m *DefaultManager) updateClientNetworks(updateSerial uint64, networks rout } config := client.WatcherConfig{ - Context: m.ctx, - DNSRouteInterval: m.dnsRouteInterval, - WGInterface: m.wgInterface, - StatusRecorder: m.statusRecorder, - Route: routes[0], - Handler: handler, + Context: m.ctx, + DNSRouteInterval: m.dnsRouteInterval, + WGInterface: m.wgInterface, + StatusRecorder: m.statusRecorder, + Route: routes[0], + Handler: handler, + ReconcileAllowedIPs: m.ReconcilePeerAllowedIPs, } clientNetworkWatcher = client.NewWatcher(config) m.clientNetworks[id] = clientNetworkWatcher