Stick new watcher creation to actual existence of af the conn

and its removal to the removal of such same conn.
Avoid debouncing and cross lock dead locking
This commit is contained in:
riccardom
2026-07-03 11:37:41 +02:00
parent f6900fb07c
commit d5a212349f
3 changed files with 37 additions and 37 deletions
+26 -7
View File
@@ -195,7 +195,6 @@ func NewConn(config ConnConfig, services ServiceDependencies) (*Conn, error) {
statusICE: worker.NewAtomicStatus(), statusICE: worker.NewAtomicStatus(),
dumpState: dumpState, dumpState: dumpState,
endpointUpdater: NewEndpointUpdater(connLog, config.WgConfig, isController(config)), endpointUpdater: NewEndpointUpdater(connLog, config.WgConfig, isController(config)),
wgWatcher: NewWGWatcher(connLog, config.WgConfig.WgInterface, config.Key, dumpState),
metricsRecorder: services.MetricsRecorder, metricsRecorder: services.MetricsRecorder,
} }
@@ -802,25 +801,43 @@ func (conn *Conn) isConnectedOnAllWay() (status guard.ConnStatus) {
}) })
} }
// enableWgWatcherIfNeeded starts a fresh watcher for the current connection. A new WGWatcher
// instance is created per attempt (rather than reusing one) so its lifecycle is bound entirely
// to conn.mu: enable/disable can never race against an old watcher goroutine's shutdown, which
// was the source of the "watcher silently fails to restart on a fast reconnect" bug. Caller must
// hold conn.mu.
func (conn *Conn) enableWgWatcherIfNeeded(enabledTime time.Time) { func (conn *Conn) enableWgWatcherIfNeeded(enabledTime time.Time) {
if !conn.wgWatcher.PrepareInitialHandshake() { if conn.wgWatcher != nil {
// a watcher is already running for the current connection
return return
} }
watcher := NewWGWatcher(conn.Log, conn.config.WgConfig.WgInterface, conn.config.Key, conn.dumpState)
watcher.PrepareInitialHandshake()
wgWatcherCtx, wgWatcherCancel := context.WithCancel(conn.ctx) wgWatcherCtx, wgWatcherCancel := context.WithCancel(conn.ctx)
conn.wgWatcher = watcher
conn.wgWatcherCancel = wgWatcherCancel conn.wgWatcherCancel = wgWatcherCancel
conn.wgWatcherWg.Add(1) conn.wgWatcherWg.Add(1)
go func() { go func() {
defer conn.wgWatcherWg.Done() defer conn.wgWatcherWg.Done()
conn.wgWatcher.EnableWgWatcher(wgWatcherCtx, enabledTime, conn.onWGDisconnected, conn.onWGHandshakeSuccess) watcher.EnableWgWatcher(wgWatcherCtx, enabledTime, conn.onWGDisconnected, conn.onWGHandshakeSuccess)
}() }()
} }
// disableWgWatcherIfNeeded stops and drops the current watcher once no transport is active. It
// only signals the watcher goroutine (cancel) and clears the reference; it never waits for the
// goroutine to exit, because the watcher's own timeout path reentrantly calls back here under
// conn.mu (via onWGDisconnected), so blocking would deadlock. The cancelled goroutine drains
// harmlessly. Caller must hold conn.mu.
func (conn *Conn) disableWgWatcherIfNeeded() { func (conn *Conn) disableWgWatcherIfNeeded() {
if conn.currentConnPriority == conntype.None && conn.wgWatcherCancel != nil { if conn.currentConnPriority != conntype.None || conn.wgWatcher == nil {
conn.wgWatcherCancel() return
conn.wgWatcherCancel = nil
} }
conn.wgWatcherCancel()
conn.wgWatcher = nil
conn.wgWatcherCancel = nil
} }
func (conn *Conn) newProxy(remoteConn net.Conn) (wgproxy.Proxy, error) { func (conn *Conn) newProxy(remoteConn net.Conn) (wgproxy.Proxy, error) {
@@ -843,7 +860,9 @@ func (conn *Conn) resetEndpoint() {
return return
} }
conn.Log.Infof("reset wg endpoint") conn.Log.Infof("reset wg endpoint")
conn.wgWatcher.Reset() if conn.wgWatcher != nil {
conn.wgWatcher.Reset()
}
if err := conn.endpointUpdater.RemoveEndpointAddress(); err != nil { if err := conn.endpointUpdater.RemoveEndpointAddress(); err != nil {
conn.Log.Warnf("failed to remove endpoint address before update: %v", err) conn.Log.Warnf("failed to remove endpoint address before update: %v", err)
} }
+8 -23
View File
@@ -3,7 +3,6 @@ package peer
import ( import (
"context" "context"
"fmt" "fmt"
"sync"
"time" "time"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
@@ -24,14 +23,15 @@ type WGInterfaceStater interface {
GetStats() (map[string]configurer.WGStats, error) GetStats() (map[string]configurer.WGStats, error)
} }
// WGWatcher is single-shot: create one instance per connection attempt, run it once via
// EnableWgWatcher, then discard it. Lifecycle (start/stop) is owned by Conn under conn.mu,
// so the watcher itself keeps no "enabled" state to go stale on a fast disconnect/reconnect.
type WGWatcher struct { type WGWatcher struct {
log *log.Entry log *log.Entry
wgIfaceStater WGInterfaceStater wgIfaceStater WGInterfaceStater
peerKey string peerKey string
stateDump *stateDump stateDump *stateDump
enabled bool
muEnabled sync.Mutex
// initialHandshake is not thread-safe; never call PrepareInitialHandshake and EnableWgWatcher concurrently. // initialHandshake is not thread-safe; never call PrepareInitialHandshake and EnableWgWatcher concurrently.
initialHandshake time.Time initialHandshake time.Time
@@ -48,25 +48,14 @@ func NewWGWatcher(log *log.Entry, wgIfaceStater WGInterfaceStater, peerKey strin
} }
} }
// PrepareInitialHandshake reserves the watcher and reads the peer's current WireGuard // PrepareInitialHandshake reads the peer's current WireGuard handshake time. It must be
// handshake time. It must be called before the peer is (re)configured on the WireGuard // called before the peer is (re)configured on the WireGuard interface, so the captured
// interface, so the captured baseline reflects the state prior to this connection attempt // baseline reflects the state prior to this connection attempt instead of racing with
// instead of racing with that configuration. Returns ok=false if the watcher is already // that configuration.
// running, in which case EnableWgWatcher must not be called. func (w *WGWatcher) PrepareInitialHandshake() {
func (w *WGWatcher) PrepareInitialHandshake() (ok bool) {
w.muEnabled.Lock()
if w.enabled {
w.muEnabled.Unlock()
return false
}
w.log.Debugf("enable WireGuard watcher") w.log.Debugf("enable WireGuard watcher")
w.enabled = true
w.muEnabled.Unlock()
handshake, _ := w.wgState() handshake, _ := w.wgState()
w.initialHandshake = handshake w.initialHandshake = handshake
return true
} }
// EnableWgWatcher runs the WireGuard watcher loop using the handshake baseline captured by // EnableWgWatcher runs the WireGuard watcher loop using the handshake baseline captured by
@@ -74,10 +63,6 @@ func (w *WGWatcher) PrepareInitialHandshake() (ok bool) {
// for context lifecycle management. // for context lifecycle management.
func (w *WGWatcher) EnableWgWatcher(ctx context.Context, enabledTime time.Time, onDisconnectedFn func(), onHandshakeSuccessFn func(when time.Time)) { 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.periodicHandshakeCheck(ctx, onDisconnectedFn, onHandshakeSuccessFn, enabledTime, w.initialHandshake)
w.muEnabled.Lock()
w.enabled = false
w.muEnabled.Unlock()
} }
// Reset signals the watcher that the WireGuard peer has been reset and a new // Reset signals the watcher that the WireGuard peer has been reset and a new
+3 -7
View File
@@ -7,7 +7,6 @@ import (
"time" "time"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
"github.com/stretchr/testify/require"
"github.com/netbirdio/netbird/client/iface/configurer" "github.com/netbirdio/netbird/client/iface/configurer"
) )
@@ -35,8 +34,7 @@ func TestWGWatcher_EnableWgWatcher(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background()) ctx, cancel := context.WithCancel(context.Background())
defer cancel() defer cancel()
ok := watcher.PrepareInitialHandshake() watcher.PrepareInitialHandshake()
require.True(t, ok, "watcher should not be enabled yet")
onDisconnected := make(chan struct{}, 1) onDisconnected := make(chan struct{}, 1)
go watcher.EnableWgWatcher(ctx, time.Now(), func() { go watcher.EnableWgWatcher(ctx, time.Now(), func() {
@@ -66,8 +64,7 @@ func TestWGWatcher_ReEnable(t *testing.T) {
watcher := NewWGWatcher(mlog, mocWgIface, "", newStateDump("peer", mlog, &Status{})) watcher := NewWGWatcher(mlog, mocWgIface, "", newStateDump("peer", mlog, &Status{}))
ctx, cancel := context.WithCancel(context.Background()) ctx, cancel := context.WithCancel(context.Background())
ok := watcher.PrepareInitialHandshake() watcher.PrepareInitialHandshake()
require.True(t, ok, "watcher should not be enabled yet")
wg := &sync.WaitGroup{} wg := &sync.WaitGroup{}
wg.Add(1) wg.Add(1)
@@ -83,8 +80,7 @@ func TestWGWatcher_ReEnable(t *testing.T) {
ctx, cancel = context.WithCancel(context.Background()) ctx, cancel = context.WithCancel(context.Background())
defer cancel() defer cancel()
ok = watcher.PrepareInitialHandshake() watcher.PrepareInitialHandshake()
require.True(t, ok, "watcher should be re-enabled after the previous run stopped")
onDisconnected := make(chan struct{}, 1) onDisconnected := make(chan struct{}, 1)
go watcher.EnableWgWatcher(ctx, time.Now(), func() { go watcher.EnableWgWatcher(ctx, time.Now(), func() {