sendnrw/netbird
2
0
Fork 0
mirror of https://github.com/netbirdio/netbird.git synced 2026-10-23 13:59:09 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main' into embedded-vnc

Browse Source 
# Conflicts:
#	client/cmd/up.go
#	client/internal/engine.go
#	client/internal/profilemanager/config.go
#	client/proto/daemon.pb.go
#	client/server/mdm.go
#	client/server/server.go
#	client/status/status.go
#	client/ui/main.go
#	client/ui/services/windowmanager.go
#	management/internals/shared/grpc/conversion.go
#	management/server/store/sql_store.go
#	management/server/store/sql_store_test.go
#	shared/management/networkmap/envelope.go
#	shared/management/proto/management.pb.go
This commit is contained in:
Viktor Liu
2026-10-19 19:49:54 +09:00
parent 1fec7c0ded cf01711971
commit 0a17d1a9f5
804 changed files with 72029 additions and 25943 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/internal/peer/conn.go
+17 -13
View File
@@ -341,7 +341,7 @@ func (conn *Conn) Close(signalToRemote bool) {
conn.Log.Errorf("failed to remove wg endpoint: %v", err)
}
if conn.evalStatus() == StatusConnected && conn.onDisconnected != nil {
if conn.onDisconnected != nil {
conn.onDisconnected(conn.config.WgConfig.RemoteKey)
}
@@ -828,7 +828,8 @@ func (conn *Conn) evalStatus() ConnStatus {
//
// The result is a tri-state:
// - ConnStatusConnected: all available transports are up
// - ConnStatusPartiallyConnected: relay is up but ICE is still pending/reconnecting
// - ConnStatusPartiallyConnected: one transport carries the traffic and the other does
// not: relay up with ICE down, or ICE up with the shared relay transport down
// - ConnStatusDisconnected: no working transport
func (conn *Conn) isConnectedOnAllWay() (status guard.ConnStatus) {
defer func() {
@@ -845,13 +846,14 @@ func (conn *Conn) isConnectedOnAllWay() (status guard.ConnStatus) {
}
return evalConnStatus(connStatusInputs{
forceRelay: IsForceRelayed(),
peerUsesRelay: conn.workerRelay.IsRelayConnectionSupportedWithPeer(),
relayConnected: conn.statusRelay.Get() == worker.StatusConnected,
remoteSupportsICE: conn.handshaker.RemoteICESupported(),
iceWorkerCreated: iceWorkerCreated,
iceStatusConnecting: conn.statusICE.Get() != worker.StatusDisconnected,
iceInProgress: iceInProgress,
forceRelay: IsForceRelayed(),
peerUsesRelay: conn.workerRelay.IsRelayConnectionSupportedWithPeer(),
relayConnected: conn.statusRelay.Get() == worker.StatusConnected,
relayTransportConnected: conn.workerRelay.IsTransportConnected(),
remoteSupportsICE: conn.handshaker.RemoteICESupported(),
iceWorkerCreated: iceWorkerCreated,
iceStatusConnected: conn.statusICE.Get() == worker.StatusConnected,
iceInProgress: iceInProgress,
})
}
@@ -1060,19 +1062,21 @@ func evalConnStatus(in connStatusInputs) guard.ConnStatus {
return boolToConnStatus(relayUsedAndUp)
}
// ICE counts as "up" when the status is anything other than Disconnected, OR
// when a negotiation is currently in progress (so we don't spam offers while one is in flight).
iceUp := in.iceStatusConnecting || in.iceInProgress
// ICE counts as "running" when either connected or attempting to connect.
iceRunning := in.iceStatusConnected || in.iceInProgress
// Relay side is acceptable if the peer doesn't rely on relay, or relay is connected.
relayOK := !in.peerUsesRelay || in.relayConnected
switch {
case iceUp && relayOK:
case iceRunning && relayOK:
return guard.ConnStatusConnected
case relayUsedAndUp:
// Relay is up but ICE is down — partially connected.
return guard.ConnStatusPartiallyConnected
case in.iceStatusConnected && !in.relayTransportConnected:
// ICE is up and the shared relay transport is down — offers cannot restore it.
return guard.ConnStatusPartiallyConnected
default:
return guard.ConnStatusDisconnected
}
client/internal/peer/conn_status.go
+8 -7
View File
@@ -17,13 +17,14 @@ const (
// tri-state connection classification. Extracted so the decision logic can be unit-tested
// without constructing full Worker/Handshaker objects.
type connStatusInputs struct {
forceRelay bool // NB_FORCE_RELAY or JS/WASM
peerUsesRelay bool // remote peer advertises relay support AND local has relay
relayConnected bool // statusRelay reports Connected (independent of whether peer uses relay)
remoteSupportsICE bool // remote peer sent ICE credentials
iceWorkerCreated bool // local WorkerICE exists (false in force-relay mode)
iceStatusConnecting bool // statusICE is anything other than Disconnected
iceInProgress bool // a negotiation is currently in flight
forceRelay bool // NB_FORCE_RELAY or JS/WASM
peerUsesRelay bool // remote peer advertises relay support AND local has relay
relayConnected bool // statusRelay reports Connected (independent of whether peer uses relay)
relayTransportConnected bool // the relay transport shared by all peers on that server is up
remoteSupportsICE bool // remote peer sent ICE credentials
iceWorkerCreated bool // local WorkerICE exists (false in force-relay mode)
iceStatusConnected bool // statusICE reports Connected
iceInProgress bool // a negotiation is currently in flight
}
// ConnStatus describe the status of a peer's connection
client/internal/peer/conn_status_eval_test.go
+71 -12
View File
@@ -30,6 +30,21 @@ func TestEvalConnStatus_ForceRelay(t *testing.T) {
},
want: guard.ConnStatusDisconnected,
},
{
name: "force relay, relay up but the shared transport reports down",
in: connStatusInputs{
forceRelay: true,
peerUsesRelay: true,
relayConnected: true,
relayTransportConnected: false,
// The ICE inputs are set so that the force-relay return is the only branch
// that can produce Connected here: without it the peer would fall through to
// relayUsedAndUp and report PartiallyConnected.
remoteSupportsICE: true,
iceWorkerCreated: true,
},
want: guard.ConnStatusConnected,
},
{
name: "force relay, peer does NOT use relay - disconnected forever",
in: connStatusInputs{
@@ -123,24 +138,28 @@ func TestEvalConnStatus_FullyAvailable(t *testing.T) {
mutator: func(in *connStatusInputs) {
in.peerUsesRelay = true
in.relayConnected = true
in.iceStatusConnecting = true
in.relayTransportConnected = true
in.iceStatusConnected = true
},
want: guard.ConnStatusConnected,
},
{
name: "ICE connected, peer does NOT use relay",
name: "ICE connected, peer does NOT use relay, shared transport down",
mutator: func(in *connStatusInputs) {
in.peerUsesRelay = false
in.relayConnected = false
in.iceStatusConnecting = true
in.relayTransportConnected = false
in.iceStatusConnected = true
},
// A peer that does not rely on relay is unaffected by the shared transport:
// relayOK is true, so the first arm matches before the transport is considered.
want: guard.ConnStatusConnected,
},
{
name: "ICE InProgress only, peer does NOT use relay",
mutator: func(in *connStatusInputs) {
in.peerUsesRelay = false
in.iceStatusConnecting = false
in.iceStatusConnected = false
in.iceInProgress = true
},
want: guard.ConnStatusConnected,
@@ -150,7 +169,8 @@ func TestEvalConnStatus_FullyAvailable(t *testing.T) {
mutator: func(in *connStatusInputs) {
in.peerUsesRelay = true
in.relayConnected = true
in.iceStatusConnecting = false
in.relayTransportConnected = true
in.iceStatusConnected = false
in.iceInProgress = false
},
want: guard.ConnStatusPartiallyConnected,
@@ -160,21 +180,60 @@ func TestEvalConnStatus_FullyAvailable(t *testing.T) {
mutator: func(in *connStatusInputs) {
in.peerUsesRelay = false
in.relayConnected = false
in.iceStatusConnecting = false
in.iceStatusConnected = false
in.iceInProgress = false
},
want: guard.ConnStatusDisconnected,
},
{
name: "ICE up, peer uses relay but relay down -> partial (relay required, ICE ignored)",
name: "ICE connected, relay down for this peer but the shared transport is up -> disconnected",
mutator: func(in *connStatusInputs) {
in.peerUsesRelay = true
in.relayConnected = false
in.iceStatusConnecting = true
in.relayTransportConnected = true
in.iceStatusConnected = true
},
// The transport is fine, so the peer itself is unreachable over relay: it may have
// moved to another server, and only an offer carries its new relay address.
want: guard.ConnStatusDisconnected,
},
{
name: "ICE connected, the shared relay transport is down -> partial",
mutator: func(in *connStatusInputs) {
in.peerUsesRelay = true
in.relayConnected = false
in.relayTransportConnected = false
in.iceStatusConnected = true
},
// ICE carries the traffic and the relay transport is restored by the relay client's
// own guard, not by offers, so this must not trigger the aggressive retry.
want: guard.ConnStatusPartiallyConnected,
},
{
name: "ICE only negotiating while the shared relay transport is down -> disconnected",
mutator: func(in *connStatusInputs) {
in.peerUsesRelay = true
in.relayConnected = false
in.relayTransportConnected = false
in.iceStatusConnected = false
in.iceInProgress = true
},
// A negotiation in flight is not a working transport, so this peer has no path at
// all and must keep the aggressive retry. Calling it partially connected spends the
// ICE retry budget and parks the guard on the hourly ticker, and nothing wakes it
// when the negotiation then fails: onICEStateDisconnected is only reached once ICE
// has reached Connected (worker_ice.go onConnectionStateChange).
want: guard.ConnStatusDisconnected,
},
{
name: "ICE down and the shared relay transport is down -> disconnected",
mutator: func(in *connStatusInputs) {
in.peerUsesRelay = true
in.relayConnected = false
in.relayTransportConnected = false
in.iceStatusConnected = false
in.iceInProgress = false
},
// relayOK = false (peer uses relay but it's down), iceUp = true
// first switch arm fails (relayOK false), relayUsedAndUp = false (relay down),
// falls into default: Disconnected.
want: guard.ConnStatusDisconnected,
},
{
@@ -182,7 +241,7 @@ func TestEvalConnStatus_FullyAvailable(t *testing.T) {
mutator: func(in *connStatusInputs) {
in.peerUsesRelay = false
in.relayConnected = true // not actually used since peer doesn't rely on it
in.iceStatusConnecting = false
in.iceStatusConnected = false
in.iceInProgress = false
},
want: guard.ConnStatusDisconnected,
client/internal/peer/conn_test.go
+44 -1
View File
@@ -3,19 +3,26 @@ package peer
import (
"context"
"fmt"
"net"
"net/netip"
"os"
"testing"
"time"
log "github.com/sirupsen/logrus"
"github.com/stretchr/testify/assert"
"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
"github.com/netbirdio/netbird/client/iface"
"github.com/netbirdio/netbird/client/iface/configurer"
"github.com/netbirdio/netbird/client/iface/wgaddr"
"github.com/netbirdio/netbird/client/iface/wgproxy"
"github.com/netbirdio/netbird/client/internal/metrics"
"github.com/netbirdio/netbird/client/internal/peer/conntype"
"github.com/netbirdio/netbird/client/internal/peer/dispatcher"
"github.com/netbirdio/netbird/client/internal/peer/guard"
"github.com/netbirdio/netbird/client/internal/peer/ice"
"github.com/netbirdio/netbird/client/internal/peer/worker"
"github.com/netbirdio/netbird/client/internal/stdnet"
"github.com/netbirdio/netbird/util"
)
@@ -42,7 +49,7 @@ func TestNewConn_interfaceFilter(t *testing.T) {
ignore := []string{iface.WgInterfaceDefault, "tun0", "zt", "ZeroTier", "utun", "wg", "ts",
"Tailscale", "tailscale"}
filter := stdnet.InterfaceFilter(ignore)
filter := stdnet.InterfaceFilter(ignore, nil)
for _, s := range ignore {
assert.Equal(t, filter(s), false)
@@ -418,3 +425,39 @@ func TestMetricsConnType_RelayedMatchesIsRelayed(t *testing.T) {
"priority %s: isRelayed and the %q metric tag must agree", priority, tag)
}
}
type stubWGIface struct{}
func (s *stubWGIface) UpdatePeer(peerKey string, allowedIps []netip.Prefix, keepAlive time.Duration, endpoint *net.UDPAddr, preSharedKey *wgtypes.Key) error {
return nil
}
func (s *stubWGIface) RemovePeer(peerKey string) error { return nil }
func (s *stubWGIface) GetStats() (map[string]configurer.WGStats, error) {
return map[string]configurer.WGStats{}, nil
}
func (s *stubWGIface) GetProxy() wgproxy.Proxy { return nil }
func (s *stubWGIface) Address() wgaddr.Address { return wgaddr.Address{} }
func (s *stubWGIface) RemoveEndpointAddress(key string) error { return nil }
func TestConn_Close_CallsOnDisconnected(t *testing.T) {
var disconnected []string
conn := newWGTimeoutTestConn(true, &disconnected)
conn.endpointUpdater = NewEndpointUpdater(conn.Log, conn.config.WgConfig, false)
conn.config.WgConfig.WgInterface = &stubWGIface{}
conn.endpointUpdater.wgConfig.WgInterface = &stubWGIface{}
conn.workerRelay = &WorkerRelay{}
ctx, cancel := context.WithCancel(context.Background())
conn.ctx = ctx
conn.ctxCancel = cancel
conn.opened = true
conn.statusICE = worker.NewAtomicStatus()
conn.statusRelay = worker.NewAtomicStatus()
conn.statusRecorder = NewRecorder("https://mgm")
// The connection is closed while in a disconnected or connecting state
// (e.g. peer dropped, ICE failed, or management disconnect).
conn.Close(false)
assert.Equal(t, []string{conn.config.WgConfig.RemoteKey}, disconnected,
"Close must notify onDisconnected so Rosenpass cleans up its peer worker")
}
client/internal/peer/guard/guard.go
+5 -3
View File
@@ -14,7 +14,8 @@ type ConnStatus int
const (
// ConnStatusDisconnected means neither ICE nor Relay is connected.
ConnStatusDisconnected ConnStatus = iota
// ConnStatusPartiallyConnected means Relay is connected but ICE is not.
// ConnStatusPartiallyConnected means one transport is usable and the other is not:
// relay connected with ICE down, or ICE connected with the shared relay transport down.
ConnStatusPartiallyConnected
// ConnStatusConnected means all required connections are established.
ConnStatusConnected
@@ -87,8 +88,9 @@ func (g *Guard) SetICEConnDisconnected() {
// - Connected: no action, the peer is fully reachable.
// - Disconnected (neither ICE nor Relay): retries aggressively with exponential backoff (800ms doubling
// up to timeout), never gives up. This ensures rapid recovery when the peer has no connectivity at all.
// - PartiallyConnected (Relay up, ICE not): retries up to 3 times with exponential backoff, then switches
// to one attempt per hour. This limits signaling traffic when relay already provides connectivity.
// - PartiallyConnected (one transport usable, the other not): retries up to 3 times
// with exponential backoff, then switches to one attempt per hour. This limits
// signaling traffic while the peer still has a working path.
//
// External events (relay/ICE disconnect, signal/relay reconnect, candidate changes) reset the retry
// counter and backoff ticker, giving ICE a fresh chance after network conditions change.
client/internal/peer/guard/sr_watcher.go
+12 -4
View File
@@ -54,8 +54,8 @@ func (w *SRWatcher) Start(disableICEMonitor bool) {
iceMonitor := NewICEMonitor(w.iFaceDiscover, w.iceConfig, GetICEMonitorPeriod())
go iceMonitor.Start(ctx, w.onICEChanged)
}
w.signalClient.SetOnReconnectedListener(w.onReconnected)
w.relayManager.SetOnReconnectedListener(w.onReconnected)
w.signalClient.SetOnReconnectedListener(w.onSignalReconnected)
w.relayManager.SetOnReconnectedListener(w.onRelayReconnected)
}
@@ -96,7 +96,15 @@ func (w *SRWatcher) onICEChanged() {
w.notify()
}
func (w *SRWatcher) onReconnected() {
func (w *SRWatcher) onSignalReconnected() {
w.onReconnected("Signal")
}
func (w *SRWatcher) onRelayReconnected() {
w.onReconnected("Relay")
}
func (w *SRWatcher) onReconnected(server string) {
if !w.signalClient.Ready() {
return
}
@@ -104,7 +112,7 @@ func (w *SRWatcher) onReconnected() {
return
}
log.Infof("reconnected to Signal or Relay server")
log.Infof("Signal and Relay ready (triggered by %s), resetting peer reconnection backoff", server)
w.notify()
}
client/internal/peer/ice/agent.go
+2 -5
View File
@@ -39,10 +39,7 @@ func NewAgent(ctx context.Context, iFaceDiscover stdnet.ExternalIFaceDiscover, c
iceFailedTimeout := iceFailedTimeout()
iceRelayAcceptanceMinWait := iceRelayAcceptanceMinWait()
transportNet, err := newStdNet(ctx, iFaceDiscover, config.InterfaceBlackList)
if err != nil {
log.Errorf("failed to create pion's stdnet: %s", err)
}
transportNet := newStdNet(ctx, iFaceDiscover, config.InterfaceBlackList, config.WGDetector)
fac := logging.NewDefaultLoggerFactory()
@@ -53,7 +50,7 @@ func NewAgent(ctx context.Context, iFaceDiscover stdnet.ExternalIFaceDiscover, c
NetworkTypes: []ice.NetworkType{ice.NetworkTypeUDP4, ice.NetworkTypeUDP6},
Urls: config.StunTurn.Load(),
CandidateTypes: candidateTypes,
InterfaceFilter: stdnet.InterfaceFilter(config.InterfaceBlackList),
InterfaceFilter: stdnet.InterfaceFilter(config.InterfaceBlackList, config.WGDetector),
UDPMux: config.UDPMux,
UDPMuxSrflx: config.UDPMuxSrflx,
NAT1To1IPs: config.NATExternalIPs,
client/internal/peer/ice/config.go
+6
View File
@@ -2,6 +2,8 @@ package ice
import (
"github.com/pion/ice/v4"
"github.com/netbirdio/netbird/client/internal/stdnet"
)
type Config struct {
@@ -17,4 +19,8 @@ type Config struct {
UDPMuxSrflx ice.UniversalUDPMux
NATExternalIPs []string
// WGDetector is shared by every agent so that the WireGuard check the interface
// filter performs is not repeated for each of them.
WGDetector *stdnet.WGDetector
}
client/internal/peer/ice/stdnet.go
+2 -2
View File
@@ -8,6 +8,6 @@ import (
"github.com/netbirdio/netbird/client/internal/stdnet"
)
func newStdNet(ctx context.Context, _ stdnet.ExternalIFaceDiscover, ifaceBlacklist []string) (*stdnet.Net, error) {
return stdnet.NewNet(ctx, ifaceBlacklist)
func newStdNet(ctx context.Context, _ stdnet.ExternalIFaceDiscover, ifaceBlacklist []string, detector *stdnet.WGDetector) *stdnet.Net {
return stdnet.NewNet(ctx, ifaceBlacklist, detector)
}
client/internal/peer/ice/stdnet_android.go
+2 -2
View File
@@ -6,6 +6,6 @@ import (
"github.com/netbirdio/netbird/client/internal/stdnet"
)
func newStdNet(ctx context.Context, iFaceDiscover stdnet.ExternalIFaceDiscover, ifaceBlacklist []string) (*stdnet.Net, error) {
return stdnet.NewNetWithDiscover(ctx, iFaceDiscover, ifaceBlacklist)
func newStdNet(ctx context.Context, iFaceDiscover stdnet.ExternalIFaceDiscover, ifaceBlacklist []string, detector *stdnet.WGDetector) *stdnet.Net {
return stdnet.NewNetWithDiscover(ctx, iFaceDiscover, ifaceBlacklist, detector)
}
client/internal/peer/notifier.go
+52 -8
View File
@@ -12,6 +12,8 @@ type notifier struct {
serverStateLock sync.Mutex
listenersLock sync.Mutex
listener Listener
peerListWake chan struct{}
peerListStop chan struct{}
currentClientState bool
lastNotification ClientState
lastNumberOfPeers int
@@ -62,7 +64,6 @@ func (n *notifier) setNetworkAvailable(available bool) {
func (n *notifier) setListener(listener Listener) {
n.serverStateLock.Lock()
lastNotification := n.effectiveState(n.lastNotification)
numOfPeers := n.lastNumberOfPeers
fqdnAddress := n.lastFqdnAddress
address := n.lastIPAddress
n.serverStateLock.Unlock()
@@ -70,17 +71,19 @@ func (n *notifier) setListener(listener Listener) {
n.listenersLock.Lock()
defer n.listenersLock.Unlock()
n.stopPeerListDelivererLocked()
n.listener = listener
listener.OnAddressChanged(fqdnAddress, address)
notifyListener(listener, lastNotification)
// run on go routine to avoid on Java layer to call go functions on same thread
go listener.OnPeersListChanged(numOfPeers)
n.startPeerListDelivererLocked(listener)
n.wakePeerListDelivererLocked()
}
func (n *notifier) removeListener() {
n.listenersLock.Lock()
defer n.listenersLock.Unlock()
n.stopPeerListDelivererLocked()
n.listener = nil
}
@@ -178,15 +181,56 @@ func (n *notifier) peerListChanged(numOfPeers int) {
n.serverStateLock.Unlock()
n.listenersLock.Lock()
listener := n.listener
n.listenersLock.Unlock()
defer n.listenersLock.Unlock()
n.wakePeerListDelivererLocked()
}
if listener == nil {
func (n *notifier) startPeerListDelivererLocked(listener Listener) {
wake := make(chan struct{}, 1)
stop := make(chan struct{})
n.peerListWake = wake
n.peerListStop = stop
go n.deliverPeerListChanges(listener, wake, stop)
}
func (n *notifier) stopPeerListDelivererLocked() {
if n.peerListStop == nil {
return
}
close(n.peerListStop)
n.peerListStop = nil
n.peerListWake = nil
}
// run on go routine to avoid on Java layer to call go functions on same thread
go listener.OnPeersListChanged(numOfPeers)
func (n *notifier) wakePeerListDelivererLocked() {
if n.peerListWake == nil {
return
}
select {
case n.peerListWake <- struct{}{}:
default:
}
}
func (n *notifier) deliverPeerListChanges(listener Listener, wake <-chan struct{}, stop <-chan struct{}) {
for {
select {
case <-stop:
return
case <-wake:
}
select {
case <-stop:
return
default:
}
n.serverStateLock.Lock()
numOfPeers := n.lastNumberOfPeers
n.serverStateLock.Unlock()
listener.OnPeersListChanged(numOfPeers)
}
}
func (n *notifier) localAddressChanged(fqdn, address string) {
client/internal/peer/notifier_test.go
+155
View File
@@ -2,7 +2,9 @@ package peer
import (
"sync"
"sync/atomic"
"testing"
"time"
)
type mocListener struct {
@@ -115,3 +117,156 @@ func Test_notifier_RemoveListener(t *testing.T) {
t.Errorf("invalid state: %d", listener.peers)
}
}
type coalescingListener struct {
final int
calls atomic.Int32
inFlight atomic.Int32
maxInFlight atomic.Int32
last atomic.Int32
done chan struct{}
entered chan struct{}
release chan struct{}
once sync.Once
}
func (l *coalescingListener) OnStateChanged(ClientState) {}
func (l *coalescingListener) OnConnected() {}
func (l *coalescingListener) OnDisconnected() {}
func (l *coalescingListener) OnConnecting() {}
func (l *coalescingListener) OnDisconnecting() {}
func (l *coalescingListener) OnAddressChanged(string, string) {}
func (l *coalescingListener) OnPeersListChanged(size int) {
current := l.inFlight.Add(1)
for {
seen := l.maxInFlight.Load()
if current <= seen || l.maxInFlight.CompareAndSwap(seen, current) {
break
}
}
if l.calls.Add(1) == 1 && l.entered != nil {
close(l.entered)
}
if l.release != nil {
<-l.release
}
time.Sleep(time.Millisecond)
l.last.Store(int32(size))
l.inFlight.Add(-1)
if size == l.final {
l.once.Do(func() { close(l.done) })
}
}
func Test_notifier_PeerListChangedCoalesces(t *testing.T) {
const events = 1000
listener := &coalescingListener{final: events, done: make(chan struct{})}
n := newNotifier()
n.setListener(listener)
for i := 1; i <= events; i++ {
n.peerListChanged(i)
}
select {
case <-listener.done:
case <-time.After(5 * time.Second):
t.Fatalf("last peer count not delivered, last seen: %d", listener.last.Load())
}
if got := listener.maxInFlight.Load(); got != 1 {
t.Errorf("concurrent deliveries: %d, expected 1", got)
}
if got := listener.calls.Load(); got >= events {
t.Errorf("deliveries not coalesced: %d calls for %d events", got, events)
}
}
func Test_notifier_SetListenerStopsPreviousDeliverer(t *testing.T) {
old := &coalescingListener{final: -1}
replacement := &coalescingListener{final: 7, done: make(chan struct{})}
n := newNotifier()
n.setListener(old)
oldStop := n.peerListStop
n.peerListChanged(7)
n.setListener(replacement)
select {
case <-oldStop:
default:
t.Fatal("old deliverer not stopped on listener replacement")
}
waitFor(t, replacement.done, "replacement listener not notified")
}
func Test_notifier_RemoveListenerStopsDeliverer(t *testing.T) {
n := newNotifier()
n.setListener(&coalescingListener{final: -1})
stop := n.peerListStop
n.removeListener()
select {
case <-stop:
default:
t.Fatal("deliverer not stopped on listener removal")
}
}
func Test_notifier_DelivererExitsAfterInFlightCallback(t *testing.T) {
listener := &coalescingListener{
final: -1,
entered: make(chan struct{}),
release: make(chan struct{}),
}
n := newNotifier()
wake := make(chan struct{}, 1)
stop := make(chan struct{})
exited := make(chan struct{})
go func() {
n.deliverPeerListChanges(listener, wake, stop)
close(exited)
}()
wake <- struct{}{}
waitFor(t, listener.entered, "listener not called")
n.peerListChanged(7)
wake <- struct{}{}
close(stop)
close(listener.release)
waitFor(t, exited, "deliverer did not exit after stop")
if got := listener.calls.Load(); got != 1 {
t.Errorf("deliverer ran %d callbacks after stop, expected only the in-flight one", got)
}
if got := listener.last.Load(); got == 7 {
t.Errorf("deliverer delivered the peer count queued after stop")
}
}
func Test_notifier_DelivererPrefersStopOverPendingWake(t *testing.T) {
listener := &coalescingListener{final: -1}
n := newNotifier()
wake := make(chan struct{}, 1)
stop := make(chan struct{})
wake <- struct{}{}
close(stop)
n.deliverPeerListChanges(listener, wake, stop)
if got := listener.calls.Load(); got != 0 {
t.Errorf("deliverer ran %d callbacks with stop closed, expected 0", got)
}
}
func waitFor(t *testing.T, ch <-chan struct{}, msg string) {
t.Helper()
select {
case <-ch:
case <-time.After(5 * time.Second):
t.Fatal(msg)
}
}
client/internal/peer/status.go
+156 -70
View File
@@ -18,9 +18,7 @@ import (
"google.golang.org/protobuf/types/known/durationpb"
"google.golang.org/protobuf/types/known/timestamppb"
firewall "github.com/netbirdio/netbird/client/firewall/manager"
"github.com/netbirdio/netbird/client/iface/configurer"
"github.com/netbirdio/netbird/client/internal/ingressgw"
"github.com/netbirdio/netbird/client/internal/relay"
"github.com/netbirdio/netbird/client/proto"
"github.com/netbirdio/netbird/route"
@@ -30,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
@@ -43,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
@@ -69,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
@@ -105,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
@@ -161,16 +183,29 @@ type FullStatus struct {
RosenpassState RosenpassState
Relays []relay.ProbeResult
NSGroupStates []NSGroupState
NumOfForwardingRules int
LazyConnectionEnabled bool
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 {
@@ -187,16 +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
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
@@ -246,8 +319,6 @@ type Status struct {
// read it without taking mux.
networksRevision atomic.Uint64
ingressGwMgr *ingressgw.Manager
routeIDLookup routeIDLookup
wgIface WGIfaceStatus
}
@@ -257,6 +328,7 @@ func NewRecorder(mgmAddress string) *Status {
return &Status{
peers: make(map[string]State),
ipToKey: make(map[string]string),
activeRoutePeers: make(map[route.HAUniqueID]string),
changeNotify: make(map[string]map[string]*StatusChangeSubscription),
eventStreams: make(map[string]chan *proto.SystemEvent),
eventQueue: NewEventQueue(eventQueueSize),
@@ -274,12 +346,6 @@ func (d *Status) SetRelayMgr(manager *relayClient.Manager) {
d.relayMgr = manager
}
func (d *Status) SetIngressGwMgr(ingressGwMgr *ingressgw.Manager) {
d.mux.Lock()
defer d.mux.Unlock()
d.ingressGwMgr = ingressGwMgr
}
// ReplaceOfflinePeers replaces
func (d *Status) ReplaceOfflinePeers(replacement []State) {
d.mux.Lock()
@@ -330,18 +396,6 @@ func (d *Status) GetPeer(peerPubKey string) (State, error) {
return state, nil
}
func (d *Status) PeerByIP(ip string) (string, bool) {
d.mux.RLock()
defer d.mux.RUnlock()
for _, state := range d.peers {
if state.IP == ip {
return state.FQDN, true
}
}
return "", false
}
// PeerStateByIP returns the full peer State for the given tunnel IP.
// Matches against either the IPv4 (State.IP) or IPv6 (State.IPv6) tunnel
// address so dual-stack peers are reachable on either family. Only
@@ -407,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
@@ -423,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
@@ -481,6 +536,24 @@ func (d *Status) RemovePeerStateRoute(peer string, route string) error {
return nil
}
func (d *Status) AddActiveRoutePeer(haID route.HAUniqueID, peer string) {
d.mux.Lock()
defer d.mux.Unlock()
d.activeRoutePeers[haID] = peer
}
func (d *Status) RemoveActiveRoutePeer(haID route.HAUniqueID) {
d.mux.Lock()
defer d.mux.Unlock()
delete(d.activeRoutePeers, haID)
}
func (d *Status) GetActiveRoutePeers() map[route.HAUniqueID]string {
d.mux.RLock()
defer d.mux.RUnlock()
return maps.Clone(d.activeRoutePeers)
}
// CheckRoutes checks if the source and destination addresses are within the same route
// and returns the resource ID of the route that contains the addresses
func (d *Status) CheckRoutes(ip netip.Addr) ([]byte, bool) {
@@ -525,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
@@ -562,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
@@ -598,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
@@ -623,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
@@ -637,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
@@ -717,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 {
@@ -731,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()
}
@@ -1060,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()
@@ -1143,16 +1237,6 @@ func (d *Status) GetRelayStates() []relay.ProbeResult {
return relayStates
}
func (d *Status) ForwardingRules() []firewall.ForwardRule {
d.mux.RLock()
defer d.mux.RUnlock()
if d.ingressGwMgr == nil {
return nil
}
return d.ingressGwMgr.Rules()
}
func (d *Status) GetDNSStates() []NSGroupState {
d.mux.RLock()
defer d.mux.RUnlock()
@@ -1187,7 +1271,6 @@ func (d *Status) GetFullStatus() FullStatus {
Relays: d.GetRelayStates(),
RosenpassState: d.GetRosenpassState(),
NSGroupStates: d.GetDNSStates(),
NumOfForwardingRules: len(d.ForwardingRules()),
LazyConnectionEnabled: d.GetLazyConnection(),
}
@@ -1243,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
}
@@ -1258,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
}
@@ -1288,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)
}
}
@@ -1571,7 +1658,6 @@ func (fs FullStatus) ToProto() *proto.FullStatus {
pbFullStatus.LocalPeerState.WgPort = int32(fs.LocalPeerState.WgPort)
pbFullStatus.LocalPeerState.RosenpassPermissive = fs.RosenpassState.Permissive
pbFullStatus.LocalPeerState.RosenpassEnabled = fs.RosenpassState.Enabled
pbFullStatus.NumberOfForwardingRules = int32(fs.NumOfForwardingRules)
pbFullStatus.LazyConnectionEnabled = fs.LazyConnectionEnabled
pbFullStatus.LocalPeerState.Networks = maps.Keys(fs.LocalPeerState.Routes)
client/internal/peer/status_test.go
+201
View File
@@ -9,6 +9,8 @@ import (
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/netbirdio/netbird/route"
)
func TestAddPeer(t *testing.T) {
@@ -42,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"
@@ -372,3 +552,24 @@ func TestMarkServerStateDoesNotNotifyWhenUnchanged(t *testing.T) {
status.MarkManagementDisconnected(err)
assert.False(t, notified(ch), "redundant disconnect should not notify")
}
func TestActiveRoutePeers(t *testing.T) {
status := NewRecorder("https://mgm")
netA := route.HAUniqueID("net-a-10.0.0.0/24")
netB := route.HAUniqueID("net-b-10.0.0.0/24")
status.AddActiveRoutePeer(netA, "peerA")
status.AddActiveRoutePeer(netB, "peerB")
active := status.GetActiveRoutePeers()
assert.Equal(t, "peerA", active[netA])
assert.Equal(t, "peerB", active[netB])
status.RemoveActiveRoutePeer(netA)
delete(active, netB)
active = status.GetActiveRoutePeers()
_, ok := active[netA]
assert.False(t, ok)
assert.Equal(t, "peerB", active[netB])
}
client/internal/peer/worker_ice.go
+159 -29
View File
@@ -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
@@ -121,11 +142,8 @@ func (w *WorkerICE) OnNewOffer(remoteOfferAnswer *OfferAnswer) {
}
}
sessionID, err := NewICESessionID()
if err != nil {
w.log.Errorf("failed to create new session ID: %s", err)
}
w.sessionID = sessionID
// Keep the ID already advertised to the remote. Answers do not get a
// reply, so changing it here makes the next offer restart both sides.
w.abandonNegotiation()
}
@@ -140,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
@@ -156,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.
@@ -205,6 +223,9 @@ func (w *WorkerICE) Close() {
w.muxAgent.Lock()
defer w.muxAgent.Unlock()
if w.agent != nil || w.agentConnecting {
w.renewSessionID()
}
if w.agent != nil {
w.agentDialerCancel()
if err := w.agent.Close(); err != nil {
@@ -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)
@@ -366,16 +394,23 @@ func (w *WorkerICE) closeAgent(agent *icemaker.ThreadSafeAgent, cancel context.C
// Only the owner of the current session may reset its state: a stale dial
// goroutine waking after a newer attempt must not clobber it.
if w.agent == agent {
sessionID, err := NewICESessionID()
if err != nil {
w.log.Errorf("failed to create new session ID: %s", err)
}
w.sessionID = sessionID
w.renewSessionID()
w.abandonNegotiation()
}
return sessionChanged
}
// renewSessionID starts a new local session, so the remote treats our next offer
// or answer as a restart. Caller holds muxAgent.
func (w *WorkerICE) renewSessionID() {
sessionID, err := NewICESessionID()
if err != nil {
w.log.Errorf("failed to create new session ID: %s", err)
return
}
w.sessionID = sessionID
}
// abandonNegotiation drops all recorded ICE session state so the worker treats the
// next offer as a fresh start instead of a duplicate of a dead negotiation. The
// agent and agentConnecting flags must change together: leaving one stale wedges
@@ -525,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) {
@@ -605,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
client/internal/peer/worker_ice_session_test.go
+375
View File
@@ -0,0 +1,375 @@
package peer
import (
"context"
"fmt"
"net"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
icemaker "github.com/netbirdio/netbird/client/internal/peer/ice"
)
func TestWorkerICE_RemoteRestartPreservesAdvertisedSession(t *testing.T) {
w := newTestWorkerICE(t)
t.Cleanup(w.Close)
w.dialFunc = parkDial
advertised := w.SessionID()
remoteSession := ICESessionID("remote-first")
offer := OfferAnswer{
IceCredentials: IceCredentials{UFrag: "remoteufrag", Pwd: "remote-password-long-enough"},
SessionID: &remoteSession,
}
w.OnNewOffer(&offer)
require.True(t, w.InProgress(), "the first remote session must start ICE")
w.muxAgent.Lock()
firstAgent := w.agent
w.muxAgent.Unlock()
// The same callback handles answers. A changed remote ID must not create
// an unannounced local ID that makes the remote restart on our next offer.
secondSession := ICESessionID("remote-restarted")
answer := offer
answer.SessionID = &secondSession
w.OnNewOffer(&answer)
assert.Equal(t, advertised, w.SessionID(), "following a remote restart must keep our advertised ID")
w.muxAgent.Lock()
secondAgent := w.agent
w.muxAgent.Unlock()
assert.NotSame(t, firstAgent, secondAgent, "the changed remote session must still rebuild ICE")
w.OnNewOffer(&answer)
w.muxAgent.Lock()
defer w.muxAgent.Unlock()
assert.Same(t, secondAgent, w.agent, "a repeated answer must keep the replacement agent")
}
func TestWorkerICE_LocalCloseChangesAdvertisedSession(t *testing.T) {
w := newTestWorkerICE(t)
dialStarted := make(chan struct{})
dialDone := make(chan struct{})
w.dialFunc = func(ctx context.Context, _ *icemaker.ThreadSafeAgent, _ *OfferAnswer) (net.Conn, error) {
close(dialStarted)
defer close(dialDone)
<-ctx.Done()
return nil, ctx.Err()
}
session := ICESessionID("remote-session")
w.OnNewOffer(&OfferAnswer{
IceCredentials: IceCredentials{UFrag: "remoteufrag", Pwd: "remote-password-long-enough"},
SessionID: &session,
})
<-dialStarted
advertised := w.SessionID()
w.Close()
assert.NotEqual(t, advertised, w.SessionID(), "a local teardown must tell the remote to restart")
closedSession := w.SessionID()
// The abandoned dial goroutine cleans up after Close returned.
<-dialDone
assert.Never(t, func() bool { return w.SessionID() != closedSession }, 200*time.Millisecond, 10*time.Millisecond,
"the late cleanup of a closed negotiation must not restart again")
w.Close()
assert.Equal(t, closedSession, w.SessionID(), "closing an idle worker must not restart again")
}
// parkDial stands in for the ICE dial. It never connects and returns once the
// negotiation is abandoned, so a test decides when a negotiation fails.
func parkDial(ctx context.Context, _ *icemaker.ThreadSafeAgent, _ *OfferAnswer) (net.Conn, error) {
<-ctx.Done()
return nil, ctx.Err()
}
func newTestSessionID(t *testing.T) ICESessionID {
t.Helper()
sid, err := NewICESessionID()
require.NoError(t, err)
return sid
}
// handshakeSide is one end of a simulated signaling exchange.
type handshakeSide interface {
// message builds the offer or answer the side would send now.
message() OfferAnswer
// receive hands a remote offer or answer to the side's ICE logic.
receive(msg OfferAnswer)
// teardowns counts negotiations the side tore down to follow a remote restart.
teardowns() int
// failAgent ends the side's current negotiation as an ICE failure does.
failAgent()
}
// workerSide drives a real WorkerICE.
type workerSide struct {
t *testing.T
w *WorkerICE
replaced int
}
func newWorkerSide(t *testing.T) *workerSide {
t.Helper()
w := newTestWorkerICE(t)
w.dialFunc = parkDial
t.Cleanup(w.Close)
return &workerSide{t: t, w: w}
}
func (s *workerSide) message() OfferAnswer {
sid := s.w.SessionID()
ufrag, pwd := s.w.GetLocalUserCredentials()
return OfferAnswer{IceCredentials: IceCredentials{UFrag: ufrag, Pwd: pwd}, SessionID: &sid}
}
func (s *workerSide) receive(msg OfferAnswer) {
before := s.agent()
s.w.OnNewOffer(&msg)
if after := s.agent(); before != nil && after != before {
s.replaced++
}
}
func (s *workerSide) teardowns() int { return s.replaced }
func (s *workerSide) agent() *icemaker.ThreadSafeAgent {
s.w.muxAgent.Lock()
defer s.w.muxAgent.Unlock()
return s.w.agent
}
// failAgent runs the cleanup the dial goroutine or the Failed state callback
// performs when the current negotiation dies.
func (s *workerSide) failAgent() {
s.t.Helper()
s.w.muxAgent.Lock()
agent, cancel := s.w.agent, s.w.agentDialerCancel
s.w.muxAgent.Unlock()
require.NotNil(s.t, agent, "failing requires a running negotiation")
s.w.closeAgent(agent, cancel)
}
// legacySide models a remote peer running a release from before this change:
// when it follows a remote restart it also picks a new session ID of its own,
// which it announces only with its next offer or answer.
type legacySide struct {
t *testing.T
sessionID ICESessionID
remoteID ICESessionID
hasAgent bool
replaced int
}
func newLegacySide(t *testing.T) *legacySide {
return &legacySide{t: t, sessionID: newTestSessionID(t)}
}
func (s *legacySide) message() OfferAnswer {
sid := s.sessionID
return OfferAnswer{
IceCredentials: IceCredentials{UFrag: "legacyufrag", Pwd: "legacy-password-long-enough"},
SessionID: &sid,
}
}
func (s *legacySide) receive(msg OfferAnswer) {
if msg.SessionID == nil {
s.hasAgent = true
return
}
if s.hasAgent {
if *msg.SessionID == s.remoteID {
return
}
s.replaced++
s.sessionID = newTestSessionID(s.t)
}
s.hasAgent = true
s.remoteID = *msg.SessionID
}
func (s *legacySide) teardowns() int { return s.replaced }
func (s *legacySide) failAgent() {
s.hasAgent = false
s.remoteID = ""
s.sessionID = newTestSessionID(s.t)
}
// exchange runs one guard-driven round in the order Handshaker.Listen uses: the
// answerer handles the offer and answers with the session ID it holds
// afterwards, and the offerer handles the answer without replying.
func exchange(offerer, answerer handshakeSide) {
answerer.receive(offerer.message())
offerer.receive(answerer.message())
}
// offerPattern decides which side's guard sends the offer in a round.
type offerPattern struct {
name string
picker func(round int, local, remote handshakeSide) (offerer, answerer handshakeSide)
}
var offerPatterns = []offerPattern{
{
// A routing peer whose relay is down keeps offering on its own.
name: "local peer offers",
picker: func(_ int, local, remote handshakeSide) (handshakeSide, handshakeSide) {
return local, remote
},
},
{
name: "both peers offer",
picker: func(round int, local, remote handshakeSide) (handshakeSide, handshakeSide) {
if round%2 == 0 {
return local, remote
}
return remote, local
},
},
}
// assertSettles runs guard rounds and requires the pair to stop restarting
// each other: at most maxTeardowns in total, and none once half the rounds ran.
func assertSettles(t *testing.T, pattern offerPattern, local, remote handshakeSide, maxTeardowns int) {
t.Helper()
const rounds = 10
total := func() int { return local.teardowns() + remote.teardowns() }
start := total()
var halfway int
for round := range rounds {
if round == rounds/2 {
halfway = total()
}
offerer, answerer := pattern.picker(round, local, remote)
exchange(offerer, answerer)
}
assert.LessOrEqual(t, total()-start, maxTeardowns, "the peers must not keep restarting each other")
assert.Equal(t, halfway, total(), "the negotiation must be stable in the later rounds")
}
// establish runs the first offer and answer, so both sides negotiate.
func establish(t *testing.T, local, remote handshakeSide) {
t.Helper()
exchange(local, remote)
require.Zero(t, local.teardowns()+remote.teardowns(), "the first exchange must not restart anything")
}
func TestICESession_SettlesAfterAgentFailure(t *testing.T) {
sides := []struct {
name string
remote func(t *testing.T) handshakeSide
}{
{name: "current remote", remote: func(t *testing.T) handshakeSide { return newWorkerSide(t) }},
{name: "legacy remote", remote: func(t *testing.T) handshakeSide { return newLegacySide(t) }},
}
failures := []struct {
name string
fail func(local, remote handshakeSide)
}{
{name: "remote agent fails", fail: func(_, remote handshakeSide) { remote.failAgent() }},
{name: "local agent fails", fail: func(local, _ handshakeSide) { local.failAgent() }},
{name: "both agents fail", fail: func(local, remote handshakeSide) {
local.failAgent()
remote.failAgent()
}},
}
for _, side := range sides {
for _, failure := range failures {
for _, pattern := range offerPatterns {
t.Run(fmt.Sprintf("%s/%s/%s", side.name, failure.name, pattern.name), func(t *testing.T) {
local := newWorkerSide(t)
remote := side.remote(t)
establish(t, local, remote)
failure.fail(local, remote)
assertSettles(t, pattern, local, remote, 2)
})
}
}
}
}
// TestICESession_LocalCloseRestartsRemote covers an explicit teardown, as on a
// WireGuard handshake timeout. The remote must start over as well, or it keeps
// answering from the negotiation this side just abandoned.
func TestICESession_LocalCloseRestartsRemote(t *testing.T) {
for _, pattern := range offerPatterns {
t.Run(pattern.name, func(t *testing.T) {
local := newWorkerSide(t)
remote := newWorkerSide(t)
establish(t, local, remote)
local.w.Close()
assertSettles(t, pattern, local, remote, 1)
assert.Equal(t, 1, remote.teardowns(), "the remote must restart its negotiation exactly once")
})
}
}
func TestICESession_DuplicateMessagesKeepNegotiation(t *testing.T) {
local := newWorkerSide(t)
remote := newWorkerSide(t)
offer := local.message()
remote.receive(offer)
answer := remote.message()
local.receive(answer)
// Signaling may deliver the same message again, and a peer answers every
// offer, including repeats of one it already handled.
remote.receive(offer)
local.receive(answer)
local.receive(remote.message())
assert.Zero(t, local.teardowns(), "a repeated answer must not restart the negotiation")
assert.Zero(t, remote.teardowns(), "a repeated offer must not restart the negotiation")
}
// TestICESession_RemoteWithoutSessionIDKeepsNegotiation covers remote peers
// too old to send session IDs: once negotiating, their messages cannot tell a
// restart from a repeat, so they must not tear anything down.
func TestICESession_RemoteWithoutSessionIDKeepsNegotiation(t *testing.T) {
local := newWorkerSide(t)
unversioned := OfferAnswer{IceCredentials: IceCredentials{UFrag: "oldufrag", Pwd: "old-password-long-enough"}}
local.receive(unversioned)
require.NotNil(t, local.agent(), "a message without a session ID must still start ICE")
advertised := local.w.SessionID()
for range 3 {
local.receive(unversioned)
}
assert.Zero(t, local.teardowns(), "messages without a session ID must not restart the negotiation")
assert.Equal(t, advertised, local.w.SessionID(), "the advertised session must not change")
}
// TestWorkerICE_StaleCleanupKeepsAdvertisedSession covers the cleanup of a
// replaced negotiation finishing late, from its dial goroutine or its Closed
// state callback. It must neither pick a new session ID, an unannounced local
// restart, nor disturb the negotiation that replaced it.
func TestWorkerICE_StaleCleanupKeepsAdvertisedSession(t *testing.T) {
local := newWorkerSide(t)
remote := newWorkerSide(t)
establish(t, local, remote)
local.w.muxAgent.Lock()
oldAgent, oldCancel := local.w.agent, local.w.agentDialerCancel
local.w.muxAgent.Unlock()
remote.failAgent()
exchange(local, remote)
require.Equal(t, 1, local.teardowns(), "the local side must follow the remote restart")
advertised := local.w.SessionID()
current := local.agent()
local.w.closeAgent(oldAgent, oldCancel)
assert.Equal(t, advertised, local.w.SessionID(), "a stale cleanup must not change the advertised session")
assert.Same(t, current, local.agent(), "a stale cleanup must keep the current negotiation")
assertSettles(t, offerPatterns[1], local, remote, 0)
}
client/internal/peer/worker_ice_test.go
+168
View File
@@ -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")
})
}
client/internal/peer/worker_relay.go
+4
View File
@@ -101,6 +101,10 @@ func (w *WorkerRelay) RelayIsSupportedLocally() bool {
return w.relayManager.HasRelayAddress()
}
func (w *WorkerRelay) IsTransportConnected() bool {
return w.relayManager.Ready()
}
func (w *WorkerRelay) CloseConn() {
w.relayLock.Lock()
conn := w.relayedConn