mirror of
https://github.com/netbirdio/netbird.git
synced 2026-10-25 22:09:08 +01:00
[client] Drop stale WG timeouts via the watcher context instead of a generation counter
Tag each WG timeout event with the watcher's context and discard it at dispatch time when that context is cancelled. This replaces the earlier generation counter and the mailbox-level epoch filter: cancelling the watcher is now the single act that both stops it and retires its pending timeouts, so the two can no longer diverge, and the staleness check moves to dispatch time so a cancel performed by an earlier event in the same drained batch already suppresses a superseded watcher's timeout.
This commit is contained in:
@@ -155,9 +155,6 @@ type Conn struct {
|
|||||||
wgWatcher *wg_watcher.WGWatcher
|
wgWatcher *wg_watcher.WGWatcher
|
||||||
wgWatcherWg sync.WaitGroup
|
wgWatcherWg sync.WaitGroup
|
||||||
wgWatcherCancel context.CancelFunc
|
wgWatcherCancel context.CancelFunc
|
||||||
// wgWatcherGen identifies the current watcher generation; a WG timeout event
|
|
||||||
// carrying an older generation is dropped by the loop. Owned by the event loop.
|
|
||||||
wgWatcherGen uint64
|
|
||||||
// wgTimeouts counts consecutive WireGuard handshake timeouts without a
|
// wgTimeouts counts consecutive WireGuard handshake timeouts without a
|
||||||
// successful handshake in between. Owned by the event loop.
|
// successful handshake in between. Owned by the event loop.
|
||||||
wgTimeouts int
|
wgTimeouts int
|
||||||
@@ -433,6 +430,10 @@ func (conn *Conn) run(mb *mailbox) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (conn *Conn) handleEvent(ev event) {
|
func (conn *Conn) handleEvent(ev event) {
|
||||||
|
if s, ok := ev.(staleableEvent); ok && s.isStale() {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
switch e := ev.(type) {
|
switch e := ev.(type) {
|
||||||
case evRemoteOffer:
|
case evRemoteOffer:
|
||||||
conn.handleRemoteOffer(&e.offer)
|
conn.handleRemoteOffer(&e.offer)
|
||||||
@@ -451,7 +452,7 @@ func (conn *Conn) handleEvent(ev event) {
|
|||||||
case evRelayDialDone:
|
case evRelayDialDone:
|
||||||
conn.handleRelayDialDone()
|
conn.handleRelayDialDone()
|
||||||
case evWGTimeout:
|
case evWGTimeout:
|
||||||
conn.handleWGTimeout(e.gen)
|
conn.handleWGTimeout()
|
||||||
case evWGHandshake:
|
case evWGHandshake:
|
||||||
conn.handleWGHandshakeSuccess(e.when)
|
conn.handleWGHandshakeSuccess(e.when)
|
||||||
case evWGCheckOK:
|
case evWGCheckOK:
|
||||||
@@ -869,16 +870,11 @@ func (conn *Conn) handleRelayDisconnected() {
|
|||||||
|
|
||||||
// handleWGTimeout closes the active connection after a WireGuard handshake
|
// handleWGTimeout closes the active connection after a WireGuard handshake
|
||||||
// timeout so the guard can trigger a reconnection.
|
// timeout so the guard can trigger a reconnection.
|
||||||
func (conn *Conn) handleWGTimeout(gen uint64) {
|
func (conn *Conn) handleWGTimeout() {
|
||||||
if conn.ctx.Err() != nil {
|
if conn.ctx.Err() != nil {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
if gen != conn.wgWatcherGen {
|
|
||||||
conn.Log.Debugf("ignore WG timeout from superseded watcher generation %d (current %d)", gen, conn.wgWatcherGen)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
conn.Log.Warnf("WireGuard handshake timeout detected, closing current connection")
|
conn.Log.Warnf("WireGuard handshake timeout detected, closing current connection")
|
||||||
|
|
||||||
// Close the active connection based on current priority
|
// Close the active connection based on current priority
|
||||||
@@ -952,8 +948,8 @@ func (conn *Conn) onGuardEvent() {
|
|||||||
conn.post(evGuardTick{})
|
conn.post(evGuardTick{})
|
||||||
}
|
}
|
||||||
|
|
||||||
func (conn *Conn) onWGDisconnected(gen uint64) {
|
func (conn *Conn) onWGDisconnected(ctx context.Context) {
|
||||||
conn.post(evWGTimeout{gen: gen})
|
conn.post(evWGTimeout{ctx: ctx})
|
||||||
}
|
}
|
||||||
|
|
||||||
func (conn *Conn) onWGHandshakeSuccess(when time.Time) {
|
func (conn *Conn) onWGHandshakeSuccess(when time.Time) {
|
||||||
@@ -1115,9 +1111,6 @@ func (conn *Conn) enableWgWatcherIfNeeded(enabledTime time.Time) {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
conn.wgWatcherGen++
|
|
||||||
gen := conn.wgWatcherGen
|
|
||||||
|
|
||||||
watcher := wg_watcher.NewWGWatcher(conn.Log, conn.config.WgConfig.WgInterface, conn.config.Key, conn.dumpState)
|
watcher := wg_watcher.NewWGWatcher(conn.Log, conn.config.WgConfig.WgInterface, conn.config.Key, conn.dumpState)
|
||||||
watcher.PrepareInitialHandshake()
|
watcher.PrepareInitialHandshake()
|
||||||
|
|
||||||
@@ -1127,7 +1120,7 @@ func (conn *Conn) enableWgWatcherIfNeeded(enabledTime time.Time) {
|
|||||||
conn.wgWatcherWg.Add(1)
|
conn.wgWatcherWg.Add(1)
|
||||||
go func() {
|
go func() {
|
||||||
defer conn.wgWatcherWg.Done()
|
defer conn.wgWatcherWg.Done()
|
||||||
onDisconnected := func() { conn.onWGDisconnected(gen) }
|
onDisconnected := func() { conn.onWGDisconnected(wgWatcherCtx) }
|
||||||
watcher.EnableWgWatcher(wgWatcherCtx, enabledTime, onDisconnected, conn.onWGHandshakeSuccess, conn.onWGCheckSuccess)
|
watcher.EnableWgWatcher(wgWatcherCtx, enabledTime, onDisconnected, conn.onWGHandshakeSuccess, conn.onWGCheckSuccess)
|
||||||
}()
|
}()
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -289,20 +289,20 @@ func TestConn_onWGDisconnected_EscalatesToRosenpassReset(t *testing.T) {
|
|||||||
conn := newWGTimeoutTestConn(true, &disconnected)
|
conn := newWGTimeoutTestConn(true, &disconnected)
|
||||||
|
|
||||||
for i := 0; i < wgTimeoutEscalationThreshold-1; i++ {
|
for i := 0; i < wgTimeoutEscalationThreshold-1; i++ {
|
||||||
conn.handleWGTimeout(conn.wgWatcherGen)
|
conn.handleWGTimeout()
|
||||||
}
|
}
|
||||||
assert.Empty(t, disconnected, "escalation must not fire below the threshold")
|
assert.Empty(t, disconnected, "escalation must not fire below the threshold")
|
||||||
|
|
||||||
conn.handleWGTimeout(conn.wgWatcherGen)
|
conn.handleWGTimeout()
|
||||||
assert.Equal(t, []string{conn.config.WgConfig.RemoteKey}, disconnected,
|
assert.Equal(t, []string{conn.config.WgConfig.RemoteKey}, disconnected,
|
||||||
"reaching the threshold must report the peer disconnected once")
|
"reaching the threshold must report the peer disconnected once")
|
||||||
|
|
||||||
for i := 0; i < wgTimeoutEscalationThreshold-1; i++ {
|
for i := 0; i < wgTimeoutEscalationThreshold-1; i++ {
|
||||||
conn.handleWGTimeout(conn.wgWatcherGen)
|
conn.handleWGTimeout()
|
||||||
}
|
}
|
||||||
assert.Len(t, disconnected, 1, "escalation must restart counting after firing")
|
assert.Len(t, disconnected, 1, "escalation must restart counting after firing")
|
||||||
|
|
||||||
conn.handleWGTimeout(conn.wgWatcherGen)
|
conn.handleWGTimeout()
|
||||||
assert.Len(t, disconnected, 2, "continued timeouts must escalate again")
|
assert.Len(t, disconnected, 2, "continued timeouts must escalate again")
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -314,12 +314,12 @@ func TestConn_onWGDisconnected_CheckSuccessResetsEscalation(t *testing.T) {
|
|||||||
conn := newWGTimeoutTestConn(true, &disconnected)
|
conn := newWGTimeoutTestConn(true, &disconnected)
|
||||||
|
|
||||||
for i := 0; i < wgTimeoutEscalationThreshold-1; i++ {
|
for i := 0; i < wgTimeoutEscalationThreshold-1; i++ {
|
||||||
conn.handleWGTimeout(conn.wgWatcherGen)
|
conn.handleWGTimeout()
|
||||||
}
|
}
|
||||||
conn.handleWGCheckSuccess()
|
conn.handleWGCheckSuccess()
|
||||||
|
|
||||||
for i := 0; i < wgTimeoutEscalationThreshold-1; i++ {
|
for i := 0; i < wgTimeoutEscalationThreshold-1; i++ {
|
||||||
conn.handleWGTimeout(conn.wgWatcherGen)
|
conn.handleWGTimeout()
|
||||||
}
|
}
|
||||||
assert.Empty(t, disconnected, "handshake success must reset the timeout count")
|
assert.Empty(t, disconnected, "handshake success must reset the timeout count")
|
||||||
}
|
}
|
||||||
@@ -327,12 +327,30 @@ func TestConn_onWGDisconnected_CheckSuccessResetsEscalation(t *testing.T) {
|
|||||||
// TestConn_onWGDisconnected_NoEscalationWithoutRosenpass: without rosenpass
|
// TestConn_onWGDisconnected_NoEscalationWithoutRosenpass: without rosenpass
|
||||||
// there is no per-peer key state to reset; repeated timeouts must not report
|
// there is no per-peer key state to reset; repeated timeouts must not report
|
||||||
// disconnects.
|
// disconnects.
|
||||||
|
func TestConn_handleEvent_DropsStaleWGTimeout(t *testing.T) {
|
||||||
|
var disconnected []string
|
||||||
|
conn := newWGTimeoutTestConn(true, &disconnected)
|
||||||
|
|
||||||
|
staleCtx, cancel := context.WithCancel(context.Background())
|
||||||
|
cancel()
|
||||||
|
for i := 0; i < wgTimeoutEscalationThreshold; i++ {
|
||||||
|
conn.handleEvent(evWGTimeout{ctx: staleCtx})
|
||||||
|
}
|
||||||
|
assert.Empty(t, disconnected, "timeouts from a cancelled watcher must be dropped")
|
||||||
|
|
||||||
|
liveCtx := context.Background()
|
||||||
|
for i := 0; i < wgTimeoutEscalationThreshold; i++ {
|
||||||
|
conn.handleEvent(evWGTimeout{ctx: liveCtx})
|
||||||
|
}
|
||||||
|
assert.Len(t, disconnected, 1, "timeouts from the live watcher must be dispatched")
|
||||||
|
}
|
||||||
|
|
||||||
func TestConn_onWGDisconnected_NoEscalationWithoutRosenpass(t *testing.T) {
|
func TestConn_onWGDisconnected_NoEscalationWithoutRosenpass(t *testing.T) {
|
||||||
var disconnected []string
|
var disconnected []string
|
||||||
conn := newWGTimeoutTestConn(false, &disconnected)
|
conn := newWGTimeoutTestConn(false, &disconnected)
|
||||||
|
|
||||||
for i := 0; i < wgTimeoutEscalationThreshold*3; i++ {
|
for i := 0; i < wgTimeoutEscalationThreshold*3; i++ {
|
||||||
conn.handleWGTimeout(conn.wgWatcherGen)
|
conn.handleWGTimeout()
|
||||||
}
|
}
|
||||||
assert.Empty(t, disconnected, "escalation must be limited to rosenpass connections")
|
assert.Empty(t, disconnected, "escalation must be limited to rosenpass connections")
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,6 +1,7 @@
|
|||||||
package peer
|
package peer
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"context"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/pion/ice/v4"
|
"github.com/pion/ice/v4"
|
||||||
@@ -15,6 +16,15 @@ import (
|
|||||||
// never mutate Conn state directly.
|
// never mutate Conn state directly.
|
||||||
type event any
|
type event any
|
||||||
|
|
||||||
|
// staleableEvent is implemented by events tied to the lifetime of a transport
|
||||||
|
// component (WG watcher, ICE agent, relay connection). Each such component runs
|
||||||
|
// under its own context, cancelled when the component is superseded; an event
|
||||||
|
// carrying a cancelled context is dropped at dispatch time. A cancel performed
|
||||||
|
// by an earlier event in the same drained batch already suppresses it.
|
||||||
|
type staleableEvent interface {
|
||||||
|
isStale() bool
|
||||||
|
}
|
||||||
|
|
||||||
// evClose asks the event loop to tear down the connection. done is closed
|
// evClose asks the event loop to tear down the connection. done is closed
|
||||||
// once the teardown finished.
|
// once the teardown finished.
|
||||||
type evClose struct {
|
type evClose struct {
|
||||||
@@ -55,9 +65,11 @@ type evRelayDown struct{}
|
|||||||
type evRelayDialDone struct{}
|
type evRelayDialDone struct{}
|
||||||
|
|
||||||
type evWGTimeout struct {
|
type evWGTimeout struct {
|
||||||
gen uint64
|
ctx context.Context
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (e evWGTimeout) isStale() bool { return e.ctx.Err() != nil }
|
||||||
|
|
||||||
// evWGHandshake reports the first WireGuard handshake of the current watcher run.
|
// evWGHandshake reports the first WireGuard handshake of the current watcher run.
|
||||||
type evWGHandshake struct {
|
type evWGHandshake struct {
|
||||||
when time.Time
|
when time.Time
|
||||||
|
|||||||
Reference in New Issue
Block a user