mirror of
https://github.com/netbirdio/netbird.git
synced 2026-07-12 01:29:56 +00:00
Compare commits
3 Commits
diagnostic
...
fix/relay_
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b053231e9a | ||
|
|
e7813bb94d | ||
|
|
12e05a586b |
@@ -803,15 +803,17 @@ func (conn *Conn) isConnectedOnAllWay() (status guard.ConnStatus) {
|
||||
}
|
||||
|
||||
func (conn *Conn) enableWgWatcherIfNeeded(enabledTime time.Time) {
|
||||
if !conn.wgWatcher.IsEnabled() {
|
||||
wgWatcherCtx, wgWatcherCancel := context.WithCancel(conn.ctx)
|
||||
conn.wgWatcherCancel = wgWatcherCancel
|
||||
conn.wgWatcherWg.Add(1)
|
||||
go func() {
|
||||
defer conn.wgWatcherWg.Done()
|
||||
conn.wgWatcher.EnableWgWatcher(wgWatcherCtx, enabledTime, conn.onWGDisconnected, conn.onWGHandshakeSuccess)
|
||||
}()
|
||||
if !conn.wgWatcher.PrepareInitialHandshake() {
|
||||
return
|
||||
}
|
||||
|
||||
wgWatcherCtx, wgWatcherCancel := context.WithCancel(conn.ctx)
|
||||
conn.wgWatcherCancel = wgWatcherCancel
|
||||
conn.wgWatcherWg.Add(1)
|
||||
go func() {
|
||||
defer conn.wgWatcherWg.Done()
|
||||
conn.wgWatcher.EnableWgWatcher(wgWatcherCtx, enabledTime, conn.onWGDisconnected, conn.onWGHandshakeSuccess)
|
||||
}()
|
||||
}
|
||||
|
||||
func (conn *Conn) disableWgWatcherIfNeeded() {
|
||||
@@ -930,22 +932,18 @@ func (conn *Conn) AgentVersionString() string {
|
||||
|
||||
func (conn *Conn) presharedKey(remoteRosenpassKey []byte) *wgtypes.Key {
|
||||
if conn.config.RosenpassConfig.PubKey == nil {
|
||||
conn.Log.Warnf("PSK-DIAG: rosenpass off -> static PSK (present=%v)", conn.config.WgConfig.PreSharedKey != nil)
|
||||
return conn.config.WgConfig.PreSharedKey
|
||||
}
|
||||
|
||||
if remoteRosenpassKey == nil && conn.config.RosenpassConfig.PermissiveMode {
|
||||
conn.Log.Warnf("PSK-DIAG: rosenpass permissive + no remote RP key -> static PSK bridge (present=%v)", conn.config.WgConfig.PreSharedKey != nil)
|
||||
return conn.config.WgConfig.PreSharedKey
|
||||
}
|
||||
|
||||
// If Rosenpass has already set a PSK for this peer, return nil to prevent
|
||||
// UpdatePeer from overwriting the Rosenpass-managed key.
|
||||
if conn.rosenpassInitializedPresharedKeyValidator != nil && conn.rosenpassInitializedPresharedKeyValidator(conn.config.Key) {
|
||||
conn.Log.Warnf("PSK-DIAG: rosenpass initialized -> returning nil (keep existing on-wire PSK; NOT re-set)")
|
||||
return nil
|
||||
}
|
||||
conn.Log.Warnf("PSK-DIAG: rosenpass strict, not yet initialized -> seeding PSK (staticPresent=%v)", conn.config.WgConfig.PreSharedKey != nil)
|
||||
|
||||
// Use NetBird PSK as the seed for Rosenpass. This same PSK is passed to
|
||||
// Rosenpass as PeerConfig.PresharedKey, ensuring the derived post-quantum
|
||||
|
||||
@@ -38,8 +38,6 @@ func (e *EndpointUpdater) ConfigureWGEndpoint(addr *net.UDPAddr, presharedKey *w
|
||||
e.mu.Lock()
|
||||
defer e.mu.Unlock()
|
||||
|
||||
e.log.Warnf("PSK-DIAG: ConfigureWGEndpoint endpoint=%s psk_present=%v", addr, presharedKey != nil)
|
||||
|
||||
if e.initiator {
|
||||
e.log.Debugf("configure up WireGuard as initiator")
|
||||
return e.configureAsInitiator(addr, presharedKey)
|
||||
@@ -53,8 +51,6 @@ func (e *EndpointUpdater) SwitchWGEndpoint(addr *net.UDPAddr, presharedKey *wgty
|
||||
e.mu.Lock()
|
||||
defer e.mu.Unlock()
|
||||
|
||||
e.log.Warnf("PSK-DIAG: SwitchWGEndpoint endpoint=%s psk_present=%v", addr, presharedKey != nil)
|
||||
|
||||
// prevent to run new update while cancel the previous update
|
||||
e.waitForCloseTheDelayedUpdate()
|
||||
|
||||
@@ -73,8 +69,6 @@ func (e *EndpointUpdater) RemoveEndpointAddress() error {
|
||||
e.mu.Lock()
|
||||
defer e.mu.Unlock()
|
||||
|
||||
e.log.Warnf("PSK-DIAG: RemoveEndpointAddress -> peer re-added with only PublicKey+AllowedIPs; on-wire PSK is dropped")
|
||||
|
||||
e.waitForCloseTheDelayedUpdate()
|
||||
return e.wgConfig.WgInterface.RemoveEndpointAddress(e.wgConfig.RemoteKey)
|
||||
}
|
||||
|
||||
@@ -31,7 +31,9 @@ type WGWatcher struct {
|
||||
stateDump *stateDump
|
||||
|
||||
enabled bool
|
||||
muEnabled sync.RWMutex
|
||||
muEnabled sync.Mutex
|
||||
// initialHandshake is not thread-safe; never call PrepareInitialHandshake and EnableWgWatcher concurrently.
|
||||
initialHandshake time.Time
|
||||
|
||||
resetCh chan struct{}
|
||||
}
|
||||
@@ -46,40 +48,38 @@ func NewWGWatcher(log *log.Entry, wgIfaceStater WGInterfaceStater, peerKey strin
|
||||
}
|
||||
}
|
||||
|
||||
// EnableWgWatcher starts the WireGuard watcher. If it is already enabled, it will return immediately and do nothing.
|
||||
// The watcher runs until ctx is cancelled. Caller is responsible for context lifecycle management.
|
||||
// NOTE: reverted to the pre-#6626 shape for bisecting the NHN issue.
|
||||
func (w *WGWatcher) EnableWgWatcher(ctx context.Context, enabledTime time.Time, onDisconnectedFn func(), onHandshakeSuccessFn func(when time.Time)) {
|
||||
// PrepareInitialHandshake reserves the watcher and reads the peer's current WireGuard
|
||||
// handshake time. It must be called before the peer is (re)configured on the WireGuard
|
||||
// interface, so the captured baseline reflects the state prior to this connection attempt
|
||||
// instead of racing with that configuration. Returns ok=false if the watcher is already
|
||||
// running, in which case EnableWgWatcher must not be called.
|
||||
func (w *WGWatcher) PrepareInitialHandshake() (ok bool) {
|
||||
w.muEnabled.Lock()
|
||||
if w.enabled {
|
||||
w.muEnabled.Unlock()
|
||||
return
|
||||
return false
|
||||
}
|
||||
|
||||
w.log.Debugf("enable WireGuard watcher")
|
||||
w.enabled = true
|
||||
w.muEnabled.Unlock()
|
||||
|
||||
initialHandshake, err := w.wgState()
|
||||
if err != nil {
|
||||
w.log.Warnf("failed to read initial wg stats: %v", err)
|
||||
}
|
||||
w.log.Warnf("PSK-DIAG: watcher baseline handshake=%v (zero=%v) [pre-6626 revert]", initialHandshake, initialHandshake.IsZero())
|
||||
handshake, _ := w.wgState()
|
||||
w.initialHandshake = handshake
|
||||
return true
|
||||
}
|
||||
|
||||
w.periodicHandshakeCheck(ctx, onDisconnectedFn, onHandshakeSuccessFn, enabledTime, initialHandshake)
|
||||
// EnableWgWatcher runs the WireGuard watcher loop using the handshake baseline captured by
|
||||
// PrepareInitialHandshake. The watcher runs until ctx is cancelled. Caller is responsible
|
||||
// for context lifecycle management.
|
||||
func (w *WGWatcher) EnableWgWatcher(ctx context.Context, enabledTime time.Time, onDisconnectedFn func(), onHandshakeSuccessFn func(when time.Time)) {
|
||||
w.periodicHandshakeCheck(ctx, onDisconnectedFn, onHandshakeSuccessFn, enabledTime, w.initialHandshake)
|
||||
|
||||
w.muEnabled.Lock()
|
||||
w.enabled = false
|
||||
w.muEnabled.Unlock()
|
||||
}
|
||||
|
||||
// IsEnabled returns true if the WireGuard watcher is currently enabled
|
||||
func (w *WGWatcher) IsEnabled() bool {
|
||||
w.muEnabled.RLock()
|
||||
defer w.muEnabled.RUnlock()
|
||||
return w.enabled
|
||||
}
|
||||
|
||||
// Reset signals the watcher that the WireGuard peer has been reset and a new
|
||||
// handshake is expected. This restarts the handshake timeout from scratch.
|
||||
func (w *WGWatcher) Reset() {
|
||||
@@ -92,7 +92,6 @@ func (w *WGWatcher) Reset() {
|
||||
// wgStateCheck help to check the state of the WireGuard handshake and relay connection
|
||||
func (w *WGWatcher) periodicHandshakeCheck(ctx context.Context, onDisconnectedFn func(), onHandshakeSuccessFn func(when time.Time), enabledTime time.Time, initialHandshake time.Time) {
|
||||
w.log.Infof("WireGuard watcher started")
|
||||
w.log.Warnf("WGW-DIAG: watcher started id=%d baseline=%v (zero=%v) firstCheckIn=%v", enabledTime.UnixNano(), initialHandshake, initialHandshake.IsZero(), wgHandshakeOvertime)
|
||||
|
||||
timer := time.NewTimer(wgHandshakeOvertime)
|
||||
defer timer.Stop()
|
||||
@@ -102,17 +101,18 @@ func (w *WGWatcher) periodicHandshakeCheck(ctx context.Context, onDisconnectedFn
|
||||
for {
|
||||
select {
|
||||
case <-timer.C:
|
||||
w.log.Warnf("WGW-DIAG: check fire id=%d lastHandshake=%v", enabledTime.UnixNano(), lastHandshake)
|
||||
handshake, ok := w.handshakeCheck(lastHandshake)
|
||||
if !ok {
|
||||
w.log.Warnf("WGW-DIAG: check failed -> firing onDisconnected (TEARDOWN, pre-6626 no ctx-recheck) id=%d", enabledTime.UnixNano())
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
onDisconnectedFn()
|
||||
return
|
||||
}
|
||||
if lastHandshake.IsZero() {
|
||||
elapsed := calcElapsed(enabledTime, *handshake)
|
||||
w.log.Infof("first wg handshake detected within: %.2fsec, (%s)", elapsed, handshake)
|
||||
if onHandshakeSuccessFn != nil {
|
||||
if onHandshakeSuccessFn != nil && ctx.Err() == nil {
|
||||
onHandshakeSuccessFn(*handshake)
|
||||
}
|
||||
}
|
||||
@@ -123,15 +123,15 @@ func (w *WGWatcher) periodicHandshakeCheck(ctx context.Context, onDisconnectedFn
|
||||
timer.Reset(resetTime)
|
||||
w.stateDump.WGcheckSuccess()
|
||||
|
||||
w.log.Warnf("WGW-DIAG: check ok id=%d handshake=%v nextCheckIn=%v", enabledTime.UnixNano(), handshake, resetTime)
|
||||
w.log.Debugf("WireGuard watcher reset timer: %v", resetTime)
|
||||
case <-w.resetCh:
|
||||
w.log.Warnf("WGW-DIAG: peer reset received, restarting timeout id=%d", enabledTime.UnixNano())
|
||||
w.log.Infof("WireGuard watcher received peer reset, restarting handshake timeout")
|
||||
lastHandshake = time.Time{}
|
||||
enabledTime = time.Now()
|
||||
timer.Stop()
|
||||
timer.Reset(wgHandshakeOvertime)
|
||||
case <-ctx.Done():
|
||||
w.log.Warnf("WGW-DIAG: watcher stopped (ctx done) id=%d", enabledTime.UnixNano())
|
||||
w.log.Infof("WireGuard watcher stopped")
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"time"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/netbirdio/netbird/client/iface/configurer"
|
||||
)
|
||||
@@ -34,6 +35,9 @@ func TestWGWatcher_EnableWgWatcher(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
ok := watcher.PrepareInitialHandshake()
|
||||
require.True(t, ok, "watcher should not be enabled yet")
|
||||
|
||||
onDisconnected := make(chan struct{}, 1)
|
||||
go watcher.EnableWgWatcher(ctx, time.Now(), func() {
|
||||
mlog.Infof("onDisconnectedFn")
|
||||
@@ -62,6 +66,9 @@ func TestWGWatcher_ReEnable(t *testing.T) {
|
||||
watcher := NewWGWatcher(mlog, mocWgIface, "", newStateDump("peer", mlog, &Status{}))
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
ok := watcher.PrepareInitialHandshake()
|
||||
require.True(t, ok, "watcher should not be enabled yet")
|
||||
|
||||
wg := &sync.WaitGroup{}
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
@@ -76,6 +83,9 @@ func TestWGWatcher_ReEnable(t *testing.T) {
|
||||
ctx, cancel = context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
ok = watcher.PrepareInitialHandshake()
|
||||
require.True(t, ok, "watcher should be re-enabled after the previous run stopped")
|
||||
|
||||
onDisconnected := make(chan struct{}, 1)
|
||||
go watcher.EnableWgWatcher(ctx, time.Now(), func() {
|
||||
onDisconnected <- struct{}{}
|
||||
|
||||
@@ -82,15 +82,9 @@ func (m *Manager) GetPubKey() []byte {
|
||||
return m.spk
|
||||
}
|
||||
|
||||
// GetAddress returns the address of the Rosenpass server.
|
||||
//
|
||||
// The server binds v4-only (0.0.0.0). Rosenpass reaches peers over their
|
||||
// WireGuard overlay IP, which is always IPv4, so a v4 socket suffices. This
|
||||
// avoids the AF_INET6 -> IPv4 send rejection (EDESTADDRREQ) that a [::]
|
||||
// (dual-stack) socket hits on macOS/BSD when sending to a 4-byte IPv4
|
||||
// destination.
|
||||
// GetAddress returns the address of the Rosenpass server
|
||||
func (m *Manager) GetAddress() *net.UDPAddr {
|
||||
return &net.UDPAddr{IP: net.IPv4zero, Port: m.port}
|
||||
return &net.UDPAddr{Port: m.port}
|
||||
}
|
||||
|
||||
// addPeer adds a new peer to the Rosenpass server
|
||||
@@ -115,14 +109,20 @@ func (m *Manager) addPeer(rosenpassPubKey []byte, rosenpassAddr string, wireGuar
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to parse rosenpass address: %w", err)
|
||||
}
|
||||
// Resolve as udp4: our Rosenpass server binds v4-only (see GetAddress)
|
||||
// and the peer WireGuard overlay IP is always IPv4. This keeps the
|
||||
// destination a 4-byte IPv4 address that matches our v4 listening
|
||||
// socket, avoiding the AF_INET6 -> IPv4 send rejection on macOS/BSD.
|
||||
peerAddr := net.JoinHostPort(wireGuardIP, strPort)
|
||||
if pcfg.Endpoint, err = net.ResolveUDPAddr("udp4", peerAddr); err != nil {
|
||||
if pcfg.Endpoint, err = net.ResolveUDPAddr("udp", peerAddr); err != nil {
|
||||
return fmt.Errorf("failed to resolve peer endpoint address: %w", err)
|
||||
}
|
||||
// Our local Rosenpass UDP server binds on the IPv6 wildcard ([::]) — see
|
||||
// GetAddress(). The remote peer's endpoint (pcfg.Endpoint) is the destination
|
||||
// our server will sendto when initiating handshakes. ResolveUDPAddr returns a
|
||||
// 4-byte IPv4 for IPv4 hosts, which the kernel rejects (EDESTADDRREQ) when
|
||||
// sent from an AF_INET6 socket. Normalize the remote endpoint to IPv4-mapped
|
||||
// IPv6 so its address family matches our listening socket.
|
||||
// TODO: maybe bind the Rosenpass UDP server to the peer wg IP addr
|
||||
if v4 := pcfg.Endpoint.IP.To4(); v4 != nil {
|
||||
pcfg.Endpoint.IP = v4.To16()
|
||||
}
|
||||
}
|
||||
peerID, err := m.server.AddPeer(pcfg)
|
||||
if err != nil {
|
||||
@@ -280,15 +280,13 @@ func (m *Manager) OnConnected(remoteWireGuardKey string, remoteRosenpassPubKey [
|
||||
}
|
||||
|
||||
rpKeyHash := hashRosenpassKey(remoteRosenpassPubKey)
|
||||
initiator := bytes.Compare(m.spk, remoteRosenpassPubKey) == 1
|
||||
log.Warnf("PSK-DIAG: remote peer %s advertises rosenpass (key %s, addr %s); starting RP negotiation as initiator=%v", remoteWireGuardKey, rpKeyHash, remoteRosenpassAddr, initiator)
|
||||
log.Debugf("received remote rosenpass key %s, my key %s", rpKeyHash, m.rpKeyHash)
|
||||
|
||||
err := m.addPeer(remoteRosenpassPubKey, remoteRosenpassAddr, wireGuardIP, remoteWireGuardKey)
|
||||
if err != nil {
|
||||
log.Errorf("failed to add rosenpass peer: %s", err)
|
||||
return
|
||||
}
|
||||
log.Warnf("PSK-DIAG: rosenpass peer added for %s (RP negotiation started; watch for 'set PSK on WG' to confirm completion)", remoteWireGuardKey)
|
||||
}
|
||||
|
||||
// IsPresharedKeyInitialized returns true if Rosenpass has completed a handshake
|
||||
|
||||
@@ -100,7 +100,6 @@ func (h *NetbirdHandler) outputKey(_ rp.KeyOutputReason, pid rp.PeerID, psk rp.K
|
||||
log.Errorf("Failed to apply rosenpass key: %v", err)
|
||||
return
|
||||
}
|
||||
log.Warnf("PSK-DIAG: rosenpass set PSK on WG for peer %s (updateOnly=%v)", peerKey, isInitialized)
|
||||
|
||||
// Mark peer as isInitialized after the successful first rotation
|
||||
if !isInitialized {
|
||||
|
||||
@@ -231,15 +231,11 @@ func (w *Watcher) getBestRouteFromStatuses(routePeerStatuses map[route.ID]router
|
||||
switch {
|
||||
case chosen == "":
|
||||
var peers []string
|
||||
for id, r := range w.routes {
|
||||
st := "unknown(no status)"
|
||||
if ps, ok := routePeerStatuses[id]; ok {
|
||||
st = fmt.Sprintf("%s relayed=%v lat=%v", ps.status, ps.relayed, ps.latency)
|
||||
}
|
||||
peers = append(peers, fmt.Sprintf("%s{%s}", r.Peer, st))
|
||||
for _, r := range w.routes {
|
||||
peers = append(peers, r.Peer)
|
||||
}
|
||||
|
||||
log.Warnf("ROUTE-DIAG: network [%v] no routing peer available yet (all candidates connecting/absent); candidates: %s", w.handler, peers)
|
||||
log.Infof("network [%v] has not been assigned a routing peer as no peers from the list %s are currently available", w.handler, peers)
|
||||
case chosen != currID:
|
||||
// we compare the current score + 10ms to the chosen score to avoid flapping between routes
|
||||
if currScore != 0 && currScore+0.01 > chosenScore {
|
||||
@@ -251,7 +247,7 @@ func (w *Watcher) getBestRouteFromStatuses(routePeerStatuses map[route.ID]router
|
||||
if rt := w.routes[chosen]; rt != nil {
|
||||
p = rt.Peer
|
||||
}
|
||||
log.Warnf("ROUTE-DIAG: new chosen route %s peer %s score %f for network [%v] (routing peer now usable)", chosen, p, chosenScore, w.handler)
|
||||
log.Infof("New chosen route is %s with peer %s with score %f for network [%v]", chosen, p, chosenScore, w.handler)
|
||||
}
|
||||
|
||||
return chosen, chosenStatus
|
||||
|
||||
@@ -30,11 +30,17 @@ type RelayTrack struct {
|
||||
relayClient *Client
|
||||
err error
|
||||
created time.Time
|
||||
// ready is closed once the dial started by openConnVia finishes, at which
|
||||
// point exactly one of relayClient/err is set. Callers that find an existing
|
||||
// track wait on this channel instead of the track lock, so the network dial
|
||||
// is never performed while holding rt.Lock(). See openConnVia.
|
||||
ready chan struct{}
|
||||
}
|
||||
|
||||
func NewRelayTrack() *RelayTrack {
|
||||
return &RelayTrack{
|
||||
created: time.Now(),
|
||||
ready: make(chan struct{}),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -294,8 +300,17 @@ func (m *Manager) RelayStates() []RelayConnState {
|
||||
|
||||
// Only connected foreign relays carry state; a failed connect is evicted
|
||||
// immediately (openConnVia), so there is no error state to surface.
|
||||
//
|
||||
// Query each track without blocking: openConnVia holds a track's write-lock
|
||||
// for the whole of relayClient.Connect() (the network dial). A blocking
|
||||
// RLock here would stall the status path (GetFullStatus -> GetRelayStates)
|
||||
// for the full dial timeout. A track mid-Connect has no relayClient set yet
|
||||
// and would be skipped anyway, so TryRLock + skip preserves the result while
|
||||
// keeping status responsive.
|
||||
for _, rt := range tracks {
|
||||
rt.RLock()
|
||||
if !rt.TryRLock() {
|
||||
continue
|
||||
}
|
||||
rc := rt.relayClient
|
||||
rt.RUnlock()
|
||||
if rc != nil {
|
||||
@@ -326,34 +341,27 @@ func (m *Manager) openConnVia(ctx context.Context, serverAddress, peerKey string
|
||||
// check if already has a connection to the desired relay server
|
||||
m.relayClientsMutex.RLock()
|
||||
rt, ok := m.relayClients[serverAddress]
|
||||
if ok {
|
||||
rt.RLock()
|
||||
m.relayClientsMutex.RUnlock()
|
||||
defer rt.RUnlock()
|
||||
if rt.err != nil {
|
||||
return nil, rt.err
|
||||
}
|
||||
return rt.relayClient.OpenConn(ctx, peerKey)
|
||||
}
|
||||
m.relayClientsMutex.RUnlock()
|
||||
if ok {
|
||||
return m.openConnOnTrack(ctx, rt, peerKey)
|
||||
}
|
||||
|
||||
// if not, establish a new connection but check it again (because changed the lock type) before starting the
|
||||
// connection
|
||||
m.relayClientsMutex.Lock()
|
||||
rt, ok = m.relayClients[serverAddress]
|
||||
if ok {
|
||||
rt.RLock()
|
||||
m.relayClientsMutex.Unlock()
|
||||
defer rt.RUnlock()
|
||||
if rt.err != nil {
|
||||
return nil, rt.err
|
||||
}
|
||||
return rt.relayClient.OpenConn(ctx, peerKey)
|
||||
return m.openConnOnTrack(ctx, rt, peerKey)
|
||||
}
|
||||
|
||||
// create a new relay client and store it in the relayClients map
|
||||
// Create the track, publish it, and release the map lock BEFORE dialing. The
|
||||
// dial must not run while holding any track lock: RelayStates() and the
|
||||
// cleanup loop take the track lock, and blocking them for the whole dial
|
||||
// timeout is what stalls `netbird status -d`. Concurrent callers find this
|
||||
// track in the map and wait on rt.ready (see openConnOnTrack), so only this
|
||||
// goroutine performs the dial and the others reuse its result.
|
||||
rt = NewRelayTrack()
|
||||
rt.Lock()
|
||||
m.relayClients[serverAddress] = rt
|
||||
m.relayClientsMutex.Unlock()
|
||||
|
||||
@@ -361,8 +369,10 @@ func (m *Manager) openConnVia(ctx context.Context, serverAddress, peerKey string
|
||||
relayClient.SetTransportFallback(m.transportFallback)
|
||||
err := relayClient.Connect(m.ctx)
|
||||
if err != nil {
|
||||
rt.Lock()
|
||||
rt.err = err
|
||||
rt.Unlock()
|
||||
close(rt.ready)
|
||||
m.relayClientsMutex.Lock()
|
||||
delete(m.relayClients, serverAddress)
|
||||
m.relayClientsMutex.Unlock()
|
||||
@@ -370,8 +380,10 @@ func (m *Manager) openConnVia(ctx context.Context, serverAddress, peerKey string
|
||||
}
|
||||
// if connection closed then delete the relay client from the list
|
||||
relayClient.SetOnDisconnectListener(m.onServerDisconnected)
|
||||
rt.Lock()
|
||||
rt.relayClient = relayClient
|
||||
rt.Unlock()
|
||||
close(rt.ready)
|
||||
|
||||
conn, err := relayClient.OpenConn(ctx, peerKey)
|
||||
if err != nil {
|
||||
@@ -380,6 +392,31 @@ func (m *Manager) openConnVia(ctx context.Context, serverAddress, peerKey string
|
||||
return conn, nil
|
||||
}
|
||||
|
||||
// openConnOnTrack opens a peer connection through an existing relay track,
|
||||
// waiting for the dial started by another openConnVia call (if still running)
|
||||
// to finish. It waits on rt.ready rather than the track lock, so it neither
|
||||
// holds nor contends the track lock across the dial; the RLock it takes
|
||||
// afterwards only guards the brief relayClient read + OpenConn, matching the
|
||||
// previous behaviour of protecting the client against a concurrent cleanup
|
||||
// close.
|
||||
func (m *Manager) openConnOnTrack(ctx context.Context, rt *RelayTrack, peerKey string) (net.Conn, error) {
|
||||
select {
|
||||
case <-rt.ready:
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
|
||||
rt.RLock()
|
||||
defer rt.RUnlock()
|
||||
if rt.err != nil {
|
||||
return nil, rt.err
|
||||
}
|
||||
if rt.relayClient == nil {
|
||||
return nil, ErrRelayClientNotConnected
|
||||
}
|
||||
return rt.relayClient.OpenConn(ctx, peerKey)
|
||||
}
|
||||
|
||||
func (m *Manager) onServerConnected() {
|
||||
m.listenerLock.Lock()
|
||||
defer m.listenerLock.Unlock()
|
||||
@@ -476,6 +513,15 @@ func (m *Manager) cleanUpUnusedRelays() {
|
||||
continue
|
||||
}
|
||||
|
||||
// The dial started by openConnVia is still in progress: the track is
|
||||
// published before Connect() completes and no longer runs under rt.Lock,
|
||||
// so relayClient is not set yet. Nothing to clean up, and it must not be
|
||||
// evicted out from under the in-flight dial.
|
||||
if rt.relayClient == nil {
|
||||
rt.Unlock()
|
||||
continue
|
||||
}
|
||||
|
||||
if time.Since(rt.created) <= m.keepUnusedServerTime {
|
||||
rt.Unlock()
|
||||
continue
|
||||
|
||||
72
shared/relay/client/manager_dial_test.go
Normal file
72
shared/relay/client/manager_dial_test.go
Normal file
@@ -0,0 +1,72 @@
|
||||
package client
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// TestManager_InProgressDialIsSafeForReadersAndCleanup covers the new state that
|
||||
// option 1 introduces: openConnVia now publishes a relay track in the map and
|
||||
// releases the map lock BEFORE dialing, so a track can legitimately exist with
|
||||
// relayClient == nil and no track lock held while its Connect() is in flight.
|
||||
//
|
||||
// The status path (RelayStates) and the cleanup loop both touch every track, so
|
||||
// both must tolerate that mid-dial state — neither deref the nil relayClient nor
|
||||
// evict a track whose dial is still running.
|
||||
func TestManager_InProgressDialIsSafeForReadersAndCleanup(t *testing.T) {
|
||||
m := &Manager{
|
||||
relayClients: make(map[string]*RelayTrack),
|
||||
// 0 so the created-time grace does not mask the mid-dial (nil relayClient)
|
||||
// guard in cleanUpUnusedRelays.
|
||||
keepUnusedServerTime: 0,
|
||||
}
|
||||
|
||||
const addr = "relay.example.com:443"
|
||||
rt := NewRelayTrack() // ready open, relayClient nil, unlocked == dial in progress
|
||||
m.relayClients[addr] = rt
|
||||
|
||||
// A status call must not block or panic on a relay still being dialed; it has
|
||||
// no state to report yet.
|
||||
require.Empty(t, m.RelayStates(), "a relay still being dialed has no state to report")
|
||||
|
||||
// Cleanup must not deref the nil relayClient, and must not evict an in-flight dial.
|
||||
m.cleanUpUnusedRelays()
|
||||
m.relayClientsMutex.RLock()
|
||||
_, stillTracked := m.relayClients[addr]
|
||||
m.relayClientsMutex.RUnlock()
|
||||
require.True(t, stillTracked, "an in-progress dial must not be cleaned up")
|
||||
}
|
||||
|
||||
// TestOpenConnOnTrack_ReleasesOnContextCancelDuringDial verifies the core option-1
|
||||
// property: a caller that finds a track whose dial is in progress waits on
|
||||
// rt.ready, not on the track lock, and can be released by its own context. This
|
||||
// is what keeps a slow relay dial from serializing behind the track lock.
|
||||
func TestOpenConnOnTrack_ReleasesOnContextCancelDuringDial(t *testing.T) {
|
||||
m := &Manager{relayClients: make(map[string]*RelayTrack)}
|
||||
|
||||
rt := NewRelayTrack() // ready deliberately left open == dial in progress
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
errCh := make(chan error, 1)
|
||||
go func() {
|
||||
_, err := m.openConnOnTrack(ctx, rt, "peerKey")
|
||||
errCh <- err
|
||||
}()
|
||||
|
||||
// While waiting for the dial the caller holds no track lock, so a concurrent
|
||||
// reader (e.g. RelayStates) can still take it.
|
||||
require.True(t, rt.TryRLock(), "waiter must not hold the track lock while the dial is in progress")
|
||||
rt.RUnlock()
|
||||
|
||||
// Caller gives up: openConnOnTrack must return via ctx rather than hang on the dial.
|
||||
cancel()
|
||||
select {
|
||||
case err := <-errCh:
|
||||
require.ErrorIs(t, err, context.Canceled)
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("openConnOnTrack did not return on ctx cancellation while a dial was in progress")
|
||||
}
|
||||
}
|
||||
62
shared/relay/client/manager_relaystates_test.go
Normal file
62
shared/relay/client/manager_relaystates_test.go
Normal file
@@ -0,0 +1,62 @@
|
||||
package client
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// TestRelayStates_DoesNotBlockWhileForeignRelayConnecting is a regression test for
|
||||
// status calls hanging behind an in-progress relay dial.
|
||||
//
|
||||
// openConnVia establishes a new foreign relay like this:
|
||||
//
|
||||
// rt = NewRelayTrack()
|
||||
// rt.Lock() // track write-lock
|
||||
// m.relayClients[serverAddress] = rt
|
||||
// m.relayClientsMutex.Unlock()
|
||||
// ...
|
||||
// err := relayClient.Connect(m.ctx) // network dial, held UNDER rt.Lock()
|
||||
// ...
|
||||
// rt.Unlock() // released only after Connect returns/times out
|
||||
//
|
||||
// So while a relay is being dialed, its RelayTrack write-lock is held for the whole
|
||||
// dial (up to serverResponseTimeout per transport attempt, times the transport
|
||||
// fallback chain, times however many relays are being dialed at once).
|
||||
//
|
||||
// RelayStates() — reached from the daemon status path via
|
||||
// peer.Status.GetFullStatus() -> GetRelayStates() -> Manager.RelayStates() — takes
|
||||
// rt.RLock() on every tracked relay. A reader lock blocks while a writer holds the
|
||||
// lock, so a single foreign relay mid-Connect stalls RelayStates(), and therefore
|
||||
// `netbird status -d`, for the full dial timeout. #6547 moved this off the shared
|
||||
// map lock but the per-track RLock still blocks the status path.
|
||||
//
|
||||
// This test recreates the exact in-progress-dial state (track present in the map
|
||||
// with its write-lock held) and asserts RelayStates() does not wait on it.
|
||||
func TestRelayStates_DoesNotBlockWhileForeignRelayConnecting(t *testing.T) {
|
||||
m := &Manager{
|
||||
relayClients: make(map[string]*RelayTrack),
|
||||
}
|
||||
|
||||
// Mirror openConnVia's state during a live dial: a track in the map whose
|
||||
// write-lock is held for the duration of relayClient.Connect().
|
||||
rt := NewRelayTrack()
|
||||
rt.Lock()
|
||||
m.relayClients["relay.example.com:443"] = rt
|
||||
// Release at the end so a (buggy) blocked RelayStates goroutine can unwind
|
||||
// instead of leaking past the test.
|
||||
t.Cleanup(rt.Unlock)
|
||||
|
||||
done := make(chan []RelayConnState, 1)
|
||||
go func() {
|
||||
done <- m.RelayStates()
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-done:
|
||||
// RelayStates returned without waiting for the in-progress dial. Good.
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("RelayStates() blocked on a relay track whose Connect() is in progress " +
|
||||
"(rt.Lock held across the dial in openConnVia); `netbird status -d` hangs for " +
|
||||
"the relay dial timeout")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user