diff --git a/client/internal/peer/conn.go b/client/internal/peer/conn.go index ee0a42299..528e9addc 100644 --- a/client/internal/peer/conn.go +++ b/client/internal/peer/conn.go @@ -155,9 +155,6 @@ type Conn struct { wgWatcher *wg_watcher.WGWatcher wgWatcherWg sync.WaitGroup 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 // successful handshake in between. Owned by the event loop. wgTimeouts int @@ -433,6 +430,10 @@ func (conn *Conn) run(mb *mailbox) { } func (conn *Conn) handleEvent(ev event) { + if s, ok := ev.(staleableEvent); ok && s.isStale() { + return + } + switch e := ev.(type) { case evRemoteOffer: conn.handleRemoteOffer(&e.offer) @@ -451,7 +452,7 @@ func (conn *Conn) handleEvent(ev event) { case evRelayDialDone: conn.handleRelayDialDone() case evWGTimeout: - conn.handleWGTimeout(e.gen) + conn.handleWGTimeout() case evWGHandshake: conn.handleWGHandshakeSuccess(e.when) case evWGCheckOK: @@ -869,16 +870,11 @@ func (conn *Conn) handleRelayDisconnected() { // handleWGTimeout closes the active connection after a WireGuard handshake // timeout so the guard can trigger a reconnection. -func (conn *Conn) handleWGTimeout(gen uint64) { +func (conn *Conn) handleWGTimeout() { if conn.ctx.Err() != nil { 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") // Close the active connection based on current priority @@ -952,8 +948,8 @@ func (conn *Conn) onGuardEvent() { conn.post(evGuardTick{}) } -func (conn *Conn) onWGDisconnected(gen uint64) { - conn.post(evWGTimeout{gen: gen}) +func (conn *Conn) onWGDisconnected(ctx context.Context) { + conn.post(evWGTimeout{ctx: ctx}) } func (conn *Conn) onWGHandshakeSuccess(when time.Time) { @@ -1115,9 +1111,6 @@ func (conn *Conn) enableWgWatcherIfNeeded(enabledTime time.Time) { return } - conn.wgWatcherGen++ - gen := conn.wgWatcherGen - watcher := wg_watcher.NewWGWatcher(conn.Log, conn.config.WgConfig.WgInterface, conn.config.Key, conn.dumpState) watcher.PrepareInitialHandshake() @@ -1127,7 +1120,7 @@ func (conn *Conn) enableWgWatcherIfNeeded(enabledTime time.Time) { conn.wgWatcherWg.Add(1) go func() { defer conn.wgWatcherWg.Done() - onDisconnected := func() { conn.onWGDisconnected(gen) } + onDisconnected := func() { conn.onWGDisconnected(wgWatcherCtx) } watcher.EnableWgWatcher(wgWatcherCtx, enabledTime, onDisconnected, conn.onWGHandshakeSuccess, conn.onWGCheckSuccess) }() } diff --git a/client/internal/peer/conn_test.go b/client/internal/peer/conn_test.go index 5ed94a1be..2b61f4470 100644 --- a/client/internal/peer/conn_test.go +++ b/client/internal/peer/conn_test.go @@ -289,20 +289,20 @@ func TestConn_onWGDisconnected_EscalatesToRosenpassReset(t *testing.T) { conn := newWGTimeoutTestConn(true, &disconnected) for i := 0; i < wgTimeoutEscalationThreshold-1; i++ { - conn.handleWGTimeout(conn.wgWatcherGen) + conn.handleWGTimeout() } 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, "reaching the threshold must report the peer disconnected once") for i := 0; i < wgTimeoutEscalationThreshold-1; i++ { - conn.handleWGTimeout(conn.wgWatcherGen) + conn.handleWGTimeout() } 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") } @@ -314,12 +314,12 @@ func TestConn_onWGDisconnected_CheckSuccessResetsEscalation(t *testing.T) { conn := newWGTimeoutTestConn(true, &disconnected) for i := 0; i < wgTimeoutEscalationThreshold-1; i++ { - conn.handleWGTimeout(conn.wgWatcherGen) + conn.handleWGTimeout() } conn.handleWGCheckSuccess() for i := 0; i < wgTimeoutEscalationThreshold-1; i++ { - conn.handleWGTimeout(conn.wgWatcherGen) + conn.handleWGTimeout() } 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 // there is no per-peer key state to reset; repeated timeouts must not report // 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) { var disconnected []string conn := newWGTimeoutTestConn(false, &disconnected) for i := 0; i < wgTimeoutEscalationThreshold*3; i++ { - conn.handleWGTimeout(conn.wgWatcherGen) + conn.handleWGTimeout() } assert.Empty(t, disconnected, "escalation must be limited to rosenpass connections") } diff --git a/client/internal/peer/event.go b/client/internal/peer/event.go index 6bfc5f8d4..c77838ff4 100644 --- a/client/internal/peer/event.go +++ b/client/internal/peer/event.go @@ -1,6 +1,7 @@ package peer import ( + "context" "time" "github.com/pion/ice/v4" @@ -15,6 +16,15 @@ import ( // never mutate Conn state directly. 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 // once the teardown finished. type evClose struct { @@ -55,9 +65,11 @@ type evRelayDown struct{} type evRelayDialDone 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. type evWGHandshake struct { when time.Time