diff --git a/handshake_manager.go b/handshake_manager.go index ff80693c..410f53b3 100644 --- a/handshake_manager.go +++ b/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) { - hh := hm.queryVpnIp(vpnIp) - if hh == nil { - return - } +func (hm *HandshakeManager) handleOutbound(hh *HandshakeHostInfo, lighthouseTriggered bool) { 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) } } diff --git a/handshake_manager_test.go b/handshake_manager_test.go index acf13539..7ecf389d 100644 --- a/handshake_manager_test.go +++ b/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 { diff --git a/timeout.go b/timeout.go index 96bf688b..a0891c6c 100644 --- a/timeout.go +++ b/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 diff --git a/timeout_test.go b/timeout_test.go index ffeecc55..b3ddfd78 100644 --- a/timeout_test.go +++ b/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)