Stop a deleted handshake's retry timers from driving the handshake that replaced it

handshake_manager.go

@@ -65,7 +65,7 @@ type HandshakeManager struct {
 	lightHouse             *LightHouse
 	outside                udp.Conn
 	config                 HandshakeConfig
-	OutboundHandshakeTimer *LockingTimerWheel[netip.Addr]
+	OutboundHandshakeTimer *LockingTimerWheel[*HandshakeHostInfo]
 	messageMetrics         *MessageMetrics
 	metricInitiated        metrics.Counter
 	metricTimedOut         metrics.Counter
@@ -81,6 +81,7 @@ type HandshakeHostInfo struct {
 
 	startTime                 time.Time        // Time that we first started trying with this handshake
 	ready                     bool             // Is the handshake ready
+	deleted                   bool             // Has the handshake been removed from the pending maps, deleters hold the lock
 	initiatingVersionOverride cert.Version     // Should we use a non-default cert version for this handshake?
 	counter                   int64            // How many attempts have we made so far
 	lastRemotes               []netip.AddrPort // Remotes that we sent to during the previous attempt
@@ -125,7 +126,7 @@ func NewHandshakeManager(l *slog.Logger, mainHostMap *HostMap, lightHouse *Light
 		outside:                outside,
 		config:                 config,
 		trigger:                make(chan netip.Addr, config.triggerBuffer),
-		OutboundHandshakeTimer: NewLockingTimerWheel[netip.Addr](config.tryInterval, hsTimeout(config.retries, config.tryInterval)),
+		OutboundHandshakeTimer: NewLockingTimerWheel[*HandshakeHostInfo](config.tryInterval, hsTimeout(config.retries, config.tryInterval)),
 		messageMetrics:         config.messageMetrics,
 		metricInitiated:        metrics.GetOrRegisterCounter("handshake_manager.initiated", nil),
 		metricTimedOut:         metrics.GetOrRegisterCounter("handshake_manager.timed_out", nil),
@@ -142,7 +143,9 @@ func (hm *HandshakeManager) Run(ctx context.Context) {
 		case <-ctx.Done():
 			return
 		case vpnIP := <-hm.trigger:
-			hm.handleOutbound(vpnIP, true)
+			if hh := hm.queryVpnIp(vpnIP); hh != nil {
+				hm.handleOutbound(hh, true)
+			}
 		case now := <-clockSource.C:
 			hm.NextOutboundHandshakeTimerTick(now)
 		}
@@ -197,23 +200,25 @@ func (hm *HandshakeManager) HandleIncoming(via ViaSender, packet []byte, h *head
 func (hm *HandshakeManager) NextOutboundHandshakeTimerTick(now time.Time) {
 	hm.OutboundHandshakeTimer.Advance(now)
 	for {
-		vpnIp, has := hm.OutboundHandshakeTimer.Purge()
+		hh, has := hm.OutboundHandshakeTimer.Purge()
 		if !has {
 			break
 		}
-		hm.handleOutbound(vpnIp, false)
+		hm.handleOutbound(hh, false)
 	}
 }
 
-func (hm *HandshakeManager) handleOutbound(vpnIp netip.Addr, lighthouseTriggered bool) {
+func (hm *HandshakeManager) handleOutbound(hh *HandshakeHostInfo, lighthouseTriggered bool) {
-	hh := hm.queryVpnIp(vpnIp)
-	if hh == nil {
-		return
-	}
 	hh.Lock()
 	defer hh.Unlock()
 
+	// Timer entries outlive their handshake. A new handshake for the same vpn addr runs its own timer chain.
+	if hh.deleted {
+		return
+	}
+
 	hostinfo := hh.hostinfo
+	vpnIp := hostinfo.vpnAddrs[0]
 	// If we are out of time, clean up
 	if hh.counter >= hm.config.retries {
 		fields := []any{
@@ -239,10 +244,14 @@ func (hm *HandshakeManager) handleOutbound(vpnIp netip.Addr, lighthouseTriggered
 	// Increment the counter to increase our delay, linear backoff
 	hh.counter++
 
+	// A lighthouse triggered attempt is still in the timer wheel, adding it again would start a second retry chain
+	if !lighthouseTriggered {
+		defer hm.OutboundHandshakeTimer.Add(hh, hm.config.tryInterval*time.Duration(hh.counter))
+	}
+
 	// Check if we have a handshake packet to transmit yet
 	if !hh.ready {
 		if !hm.buildStage0Packet(hh) {
-			hm.OutboundHandshakeTimer.Add(vpnIp, hm.config.tryInterval*time.Duration(hh.counter))
 			return
 		}
 	}
@@ -323,11 +332,6 @@ func (hm *HandshakeManager) handleOutbound(vpnIp netip.Addr, lighthouseTriggered
 	}
 
 	hm.f.relayManager.StartRelays(hm.f, vpnIp, hh, stage0)
-
-	// If a lighthouse triggered this attempt then we are still in the timer wheel and do not need to re-add
-	if !lighthouseTriggered {
-		hm.OutboundHandshakeTimer.Add(vpnIp, hm.config.tryInterval*time.Duration(hh.counter))
-	}
 }
 
 // GetOrHandshake will try to find a hostinfo with a fully formed tunnel or start a new handshake if one is not present
@@ -380,7 +384,7 @@ func (hm *HandshakeManager) StartHandshake(vpnAddr netip.Addr, cacheCb func(*Han
 	}
 	hm.vpnIps[vpnAddr] = hh
 	hm.metricInitiated.Inc(1)
-	hm.OutboundHandshakeTimer.Add(vpnAddr, hm.config.tryInterval)
+	hm.OutboundHandshakeTimer.Add(hh, hm.config.tryInterval)
 
 	if cacheCb != nil {
 		cacheCb(hh)
@@ -532,6 +536,7 @@ func (hm *HandshakeManager) DeleteHostInfo(hostinfo *HostInfo) {
 func (hm *HandshakeManager) unlockedDeleteHostInfo(hostinfo *HostInfo) {
 	for _, addr := range hostinfo.vpnAddrs {
 		if cur, ok := hm.vpnIps[addr]; ok && cur.hostinfo == hostinfo {
+			cur.deleted = true
 			delete(hm.vpnIps, addr)
 		}
 	}

handshake_manager_test.go

@@ -65,7 +65,69 @@ func Test_NewHandshakeManagerVpnIp(t *testing.T) {
 	assert.NotContains(t, blah.vpnIps, ip)
 }
 
-func testCountTimerWheelEntries(tw *LockingTimerWheel[netip.Addr]) (c int) {
+// newTestHandshakeManager returns a manager whose handshakes never build a stage 0 packet (the cert has no key), so
+// every attempt just bumps the counter and schedules the next one. tick advances its timer wheel by one tryInterval.
+func newTestHandshakeManager() (hm *HandshakeManager, tick func()) {
+	l := test.NewLogger()
+	mainHM := newHostMap(l)
+	mainHM.preferredRanges.Store(&[]netip.Prefix{})
+
+	hm = NewHandshakeManager(l, mainHM, newTestLighthouse(), &udp.NoopConn{}, defaultHandshakeConfig)
+	hm.f = &Interface{handshakeManager: hm, pki: &PKI{}, l: l}
+	hm.f.pki.cs.Store(&CertState{
+		initiatingVersion: cert.Version1,
+		privateKey:        []byte{},
+		v1Cert:            &dummyCert{version: cert.Version1},
+	})
+
+	now := time.Now()
+	hm.NextOutboundHandshakeTimerTick(now)
+	return hm, func() {
+		now = now.Add(DefaultHandshakeTryInterval)
+		hm.NextOutboundHandshakeTimerTick(now)
+	}
+}
+
+func Test_HandshakeManager_RestartDropsOldTimer(t *testing.T) {
+	ip := netip.MustParseAddr("172.1.1.2")
+	hm, tick := newTestHandshakeManager()
+
+	hm.StartHandshake(ip, nil)
+	old := hm.queryVpnIp(ip)
+	tick()
+	tick()
+	require.Equal(t, int64(1), old.counter)
+
+	// Restart the handshake while the old one still has a retry queued
+	hm.DeleteHostInfo(old.hostinfo)
+	hm.StartHandshake(ip, nil)
+	hh := hm.queryVpnIp(ip)
+	require.NotSame(t, old, hh)
+
+	// An entry for timeout d fires d/tryInterval+1 ticks after it is added, so linear backoff lands attempts on ticks
+	// 2, 4 and 7
+	var counters []int64
+	for range 7 {
+		tick()
+		counters = append(counters, hh.counter)
+	}
+	assert.Equal(t, []int64{0, 1, 1, 2, 2, 2, 3}, counters)
+
+	// The old handshake's queued retry ended its chain instead of retrying a deleted handshake
+	assert.Equal(t, int64(1), old.counter)
+	assert.Equal(t, 1, testCountTimerWheelEntries(hm.OutboundHandshakeTimer))
+}
+
+func Test_HandshakeManager_LighthouseTriggerKeepsOneTimer(t *testing.T) {
+	ip := netip.MustParseAddr("172.1.1.2")
+	hm, _ := newTestHandshakeManager()
+
+	hm.StartHandshake(ip, nil)
+	hm.handleOutbound(hm.queryVpnIp(ip), true)
+	assert.Equal(t, 1, testCountTimerWheelEntries(hm.OutboundHandshakeTimer))
+}
+
+func testCountTimerWheelEntries(tw *LockingTimerWheel[*HandshakeHostInfo]) (c int) {
 	for _, i := range tw.t.wheel {
 		n := i.Head
 		for n != nil {

timeout.go

@@ -147,7 +147,10 @@ func (tw *TimerWheel[T]) Purge() (T, bool) {
 		tw.expired.Tail = nil
 	}
 
-	// Clear out the items references
+	// Clear out the items references, a cached item must not keep its value alive
+	v := ti.Item
+	var zero T
+	ti.Item = zero
 	ti.Next = nil
 
 	// Maybe cache it for later
@@ -157,7 +160,7 @@ func (tw *TimerWheel[T]) Purge() (T, bool) {
 		tw.itemsCached++
 	}
 
-	return ti.Item, true
+	return v, true
 }
 
 // findWheel find the next position in the wheel for the provided timeout given the current tick

timeout_test.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/slackhq/nebula/firewall"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestNewTimerWheel(t *testing.T) {
@@ -150,7 +151,8 @@ func TestTimerWheel_Purge(t *testing.T) {
 	assert.Equal(t, 4, tw.itemsCached)
 	ci := tw.itemCache
 	for range 4 {
-		assert.NotNil(t, ci)
+		require.NotNil(t, ci)
+		assert.Zero(t, ci.Item)
 		ci = ci.Next
 	}
 	assert.Nil(t, ci)