fix(basichost): fix some bugs in addrs reachability manager

p2p/host/basic/addrs_manager.go

@@ -413,7 +413,7 @@ func (a *addrsManager) notifyAddrsUpdated(emitter event.Emitter, localAddrsEmitt
 		}
 	}
 	if areAddrsDifferent(previous.addrs, current.addrs) {
-		log.Debug("host addresses updated", "addrs", current.localAddrs)
+		log.Debug("host addresses updated", "addrs", current.addrs)
 		a.emitLocalAddrsUpdated(localAddrsEmitter, current.addrs, previous.addrs)
 	}
 

p2p/host/basic/addrs_reachability_tracker.go

@@ -349,7 +349,7 @@ const (
 	// This is used to prevent infinite probing of an address whose status is indeterminate for any reason.
 	maxRecentDialsPerAddr = 10
 	// confidence is the absolute difference between the number of successes and failures for an address
-	// targetConfidence is the confidence threshold for an address after which we wait for `maxProbeInterval`
+	// targetConfidence is the confidence threshold for an address after which we wait for `highConfidenceAddrProbeInterval`
 	// before probing again.
 	targetConfidence = 3
 	// minConfidence is the confidence threshold for an address to be considered reachable or unreachable
@@ -359,13 +359,11 @@ const (
 	//
 	// +2 allows for 1 invalid probe result. Consider a string of successes, after which we have a single failure
 	// and then a success(...S S S S F S). The confidence in the targetConfidence window  will be equal to
-	// targetConfidence, the last F and S cancel each other, and we won't probe again for maxProbeInterval.
+	// targetConfidence, the last F and S cancel each other, and we won't probe again for highConfidenceAddrProbeInterval.
 	maxRecentDialsWindow = targetConfidence + 2
 	// highConfidenceAddrProbeInterval is the maximum interval between probes for an address
 	highConfidenceAddrProbeInterval = 1 * time.Hour
-	// highConfidenceSecondaryAddrProbeInterval is the maximum interval between probes for an address
-	highConfidenceSecondaryAddrProbeInterval = 3 * time.Hour
-	// maxProbeResultTTL is the maximum time to keep probe results for a primary address
+	// maxProbeResultTTL is the maximum time to keep probe results for an address
 	maxProbeResultTTL = maxRecentDialsWindow * highConfidenceAddrProbeInterval
 )
 
@@ -665,8 +663,7 @@ func (s *addrStatus) requiredProbeCountForConfirmation(now time.Time) int {
 	}
 	lastOutcome := s.outcomes[len(s.outcomes)-1]
 	// If the last probe result is old, we need to retest
-	if d := now.Sub(lastOutcome.At); (s.primary == nil && d > highConfidenceAddrProbeInterval) ||
-		(d > highConfidenceSecondaryAddrProbeInterval) {
+	if now.Sub(lastOutcome.At) > highConfidenceAddrProbeInterval {
 		return 1
 	}
 	// if the last probe result was different from reachability, probe again.
@@ -834,7 +831,7 @@ func assignPrimaryAddrs(statuses map[string]*addrStatus) {
 				score += 1
 			case ma.P_WEBTRANSPORT:
 				score += 1 << 1
-			case ma.P_WEBRTC:
+			case ma.P_WEBRTC_DIRECT:
 				score += 1 << 2
 			case ma.P_WS, ma.P_WSS:
 				score += 1 << 3

p2p/host/basic/addrs_reachability_tracker_test.go

@@ -10,7 +10,6 @@ import (
 	"net/netip"
 	"slices"
 	"strings"
-	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -254,17 +253,21 @@ func TestProbeManager(t *testing.T) {
 		matest.AssertMultiaddrsMatch(t, []ma.Multiaddr{tcp, websocket}, reachable)
 		matest.AssertMultiaddrsMatch(t, []ma.Multiaddr{quic, webrtc}, unreachable)
 
-		// After highConfidenceAddrProbeInterval (1h), only primaries need refresh.
-		// websocket inherits from tcp (Public), webrtc has longer refresh interval (3h).
+		// After highConfidenceAddrProbeInterval (1h) every probed address needs a
+		// refresh: tcp and quic (primaries) plus webrtc (a secondary with a Private
+		// primary, so it doesn't inherit and refreshes on the same cadence).
+		// websocket inherits Public from tcp and is never probed.
 		for range 2 {
 			cl.Add(highConfidenceAddrProbeInterval + 1*time.Millisecond)
 			reqs := nextProbe(pm)
-			// Only tcp and quic need refresh; websocket inherits, webrtc has 3h interval
-			matest.AssertEqualMultiaddrs(t, []ma.Multiaddr{tcp, quic}, extractAddrs(reqs))
+			matest.AssertEqualMultiaddrs(t, []ma.Multiaddr{tcp, quic, webrtc}, extractAddrs(reqs))
 			pm.CompleteProbe(reqs, autonatv2.Result{Addr: tcp, Idx: 0, Reachability: network.ReachabilityPublic}, nil)
 			reqs = nextProbe(pm)
-			matest.AssertEqualMultiaddrs(t, []ma.Multiaddr{quic}, extractAddrs(reqs))
+			matest.AssertEqualMultiaddrs(t, []ma.Multiaddr{quic, webrtc}, extractAddrs(reqs))
 			pm.CompleteProbe(reqs, autonatv2.Result{Addr: quic, Idx: 0, Reachability: network.ReachabilityPrivate}, nil)
+			reqs = nextProbe(pm)
+			matest.AssertEqualMultiaddrs(t, []ma.Multiaddr{webrtc}, extractAddrs(reqs))
+			pm.CompleteProbe(reqs, autonatv2.Result{Addr: webrtc, Idx: 0, Reachability: network.ReachabilityPrivate}, nil)
 
 			reqs = nextProbe(pm)
 			require.Empty(t, reqs)
@@ -273,24 +276,6 @@ func TestProbeManager(t *testing.T) {
 		reachable, unreachable, _ = pm.AppendConfirmedAddrs(nil, nil, nil)
 		matest.AssertMultiaddrsMatch(t, reachable, []ma.Multiaddr{tcp, websocket})
 		matest.AssertMultiaddrsMatch(t, unreachable, []ma.Multiaddr{quic, webrtc})
-
-		// After highConfidenceSecondaryAddrProbeInterval (3h), webrtc needs refresh too.
-		// We've advanced 2h+2ms, need to reach 3h+ for webrtc's refresh.
-		// Also need to exceed 1h since last tcp/quic refresh for them to need refresh.
-		cl.Add(highConfidenceSecondaryAddrProbeInterval - 2*highConfidenceAddrProbeInterval + 1*time.Millisecond)
-		reqs = nextProbe(pm)
-		// tcp, quic, and webrtc need refresh; websocket still inherits from tcp
-		matest.AssertEqualMultiaddrs(t, []ma.Multiaddr{tcp, quic, webrtc}, extractAddrs(reqs))
-		pm.CompleteProbe(reqs, autonatv2.Result{Addr: tcp, Idx: 0, Reachability: network.ReachabilityPublic}, nil)
-		reqs = nextProbe(pm)
-		matest.AssertEqualMultiaddrs(t, []ma.Multiaddr{quic, webrtc}, extractAddrs(reqs))
-		pm.CompleteProbe(reqs, autonatv2.Result{Addr: quic, Idx: 0, Reachability: network.ReachabilityPrivate}, nil)
-		reqs = nextProbe(pm)
-		matest.AssertEqualMultiaddrs(t, []ma.Multiaddr{webrtc}, extractAddrs(reqs))
-		pm.CompleteProbe(reqs, autonatv2.Result{Addr: webrtc, Idx: 0, Reachability: network.ReachabilityPrivate}, nil)
-
-		reqs = nextProbe(pm)
-		require.Empty(t, reqs)
 	})
 }
 
@@ -596,6 +581,84 @@ func TestAddrsReachabilityTracker(t *testing.T) {
 			t.Fatal("expected probe")
 		}
 	})
+
+	t.Run("secondary addr stays confirmed across the result TTL", func(t *testing.T) {
+		// quic-v1 (primary) is unreachable, webrtc-direct (secondary sharing the
+		// socket) is reachable. Since the primary isn't Public the secondary
+		// doesn't inherit and keeps its own status, refreshed on the 1h cadence.
+		// Advancing well past maxProbeResultTTL (5h) must not drop it to Unknown:
+		// before the fix the secondary refreshed only every 3h and expired out of
+		// the 5h window.
+		quic := ma.StringCast("/ip4/1.1.1.1/udp/1/quic-v1")
+		webrtc := ma.StringCast("/ip4/1.1.1.1/udp/1/webrtc-direct")
+
+		notify := make(chan struct{}, 1)
+		drainNotify := func() {
+			for {
+				select {
+				case <-notify:
+				default:
+					return
+				}
+			}
+		}
+		mockClient := mockAutoNATClient{
+			F: func(_ context.Context, reqs []autonatv2.Request) (autonatv2.Result, error) {
+				select {
+				case notify <- struct{}{}:
+				default:
+				}
+				if reqs[0].Addr.Equal(webrtc) {
+					return autonatv2.Result{Addr: webrtc, Idx: 0, Reachability: network.ReachabilityPublic}, nil
+				}
+				return autonatv2.Result{Addr: reqs[0].Addr, Idx: 0, Reachability: network.ReachabilityPrivate}, nil
+			},
+		}
+
+		cl := clock.NewMock()
+		tr := newTracker(mockClient, cl)
+		tr.UpdateAddrs([]ma.Multiaddr{quic, webrtc})
+		assertFirstEvent(t, tr, []ma.Multiaddr{quic, webrtc})
+
+		drainEvent := func() {
+			select {
+			case <-tr.reachabilityUpdateCh:
+			default:
+			}
+		}
+
+		time.Sleep(100 * time.Millisecond) // let the background goroutine process the new addrs
+		cl.Add(1)                          // fire the initial probe timer
+		time.Sleep(100 * time.Millisecond) // let the probes run
+		drainNotify()
+		reachable, unreachable, _ := tr.ConfirmedAddrs()
+		require.Equal(t, []ma.Multiaddr{webrtc}, reachable)
+		require.Equal(t, []ma.Multiaddr{quic}, unreachable)
+
+		// Refresh repeatedly past the 5h TTL. Each cycle advances just past the
+		// 1h refresh interval plus one ticker interval so a refresh fires. The
+		// confirmed reachability must not change: a flap to Unknown (before the
+		// fix) shows up as a spurious reachability-update event.
+		for range 6 {
+			drainNotify()
+			drainEvent()
+			cl.Add(highConfidenceAddrProbeInterval + defaultReachabilityRefreshInterval + time.Millisecond)
+			select {
+			case <-notify:
+			case <-time.After(1 * time.Second):
+				t.Fatal("expected a refresh probe")
+			}
+			time.Sleep(100 * time.Millisecond) // let the probe results settle
+			select {
+			case <-tr.reachabilityUpdateCh:
+				t.Fatal("unexpected reachability change: secondary flapped")
+			default:
+			}
+			reachable, _, unknown := tr.ConfirmedAddrs()
+			require.Equal(t, []ma.Multiaddr{webrtc}, reachable, "secondary must stay reachable")
+			require.Empty(t, unknown)
+		}
+	})
 }
 
 func TestRefreshReachability(t *testing.T) {
@@ -646,50 +709,65 @@ func TestRefreshReachability(t *testing.T) {
 	})
 
 	t.Run("quits on cancellation", func(t *testing.T) {
-		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-		block := make(chan struct{})
+		ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
+		defer cancel()
+
+		// An addr per worker, so every worker holds a probe and a refresh that
+		// honors cancellation reaches the client exactly maxConcurrency times.
+		// (A fresh addr admits only targetConfidence concurrent probes, so a
+		// single addr couldn't guarantee a probe for every worker.)
+		addrs := make([]ma.Multiaddr, 0, defaultMaxConcurrency)
+		for i := range defaultMaxConcurrency {
+			addrs = append(addrs, ma.StringCast(fmt.Sprintf("/ip4/1.1.1.1/tcp/%d", i+1)))
+		}
+
+		inFlight := make(chan struct{}, defaultMaxConcurrency)
+		var probes atomic.Int32
 		mockClient := mockAutoNATClient{
-			F: func(_ context.Context, _ []autonatv2.Request) (autonatv2.Result, error) {
-				block <- struct{}{}
-				return autonatv2.Result{}, nil
+			F: func(ctx context.Context, _ []autonatv2.Request) (autonatv2.Result, error) {
+				// Checked on the test goroutine below. Failing here instead would
+				// risk logging to a t that has already completed.
+				if probes.Add(1) > defaultMaxConcurrency {
+					return autonatv2.Result{}, autonatv2.ErrNoPeers // persistent: unwinds the workers
+				}
+				inFlight <- struct{}{}
+				<-ctx.Done() // keep the probe in flight until it's cancelled
+				return autonatv2.Result{}, ctx.Err()
 			},
 		}
 
 		pm := newProbeManager(time.Now)
-		pm.UpdateAddrs([]ma.Multiaddr{pub1})
+		pm.UpdateAddrs(addrs)
 		r := &addrsReachabilityTracker{
-			ctx:          ctx,
-			cancel:       cancel,
-			client:       mockClient,
-			probeManager: pm,
-			clock:        clock.New(),
-		}
-		var wg sync.WaitGroup
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-			result := r.refreshReachability()
-			assert.False(t, <-result.BackoffCh)
-			assert.Equal(t, pm.InProgressProbes(), 0)
-		}()
-
-		cancel()
-		time.Sleep(50 * time.Millisecond) // wait for the cancellation to be processed
+			ctx:            ctx,
+			cancel:         cancel,
+			client:         mockClient,
+			probeManager:   pm,
+			clock:          clock.New(),
+			maxConcurrency: defaultMaxConcurrency,
+		}
 
-	outer:
+		result := r.refreshReachability()
+		// Only cancel once every worker is parked inside the client, so cancellation
+		// has to interrupt probes that are genuinely running.
 		for range defaultMaxConcurrency {
 			select {
-			case <-block:
-			default:
-				break outer
+			case <-inFlight:
+			case <-time.After(5 * time.Second):
+				t.Fatal("expected every worker to have a probe in flight")
 			}
 		}
+		cancel()
+
 		select {
-		case <-block:
-			t.Fatal("expected no more requests")
-		case <-time.After(50 * time.Millisecond):
-		}
-		wg.Wait()
+		case backoff := <-result.BackoffCh:
+			// The workers are done, so probes is final.
+			require.Equal(t, int32(defaultMaxConcurrency), probes.Load(), "started a new probe after cancellation")
+			require.False(t, backoff)
+		case <-time.After(5 * time.Second):
+			t.Fatal("refreshReachability didn't return after cancellation")
+		}
+		require.Equal(t, 0, pm.InProgressProbes())
 	})
 
 	t.Run("handles refusals", func(t *testing.T) {

p2p/protocol/autonatv2/autonat.go

@@ -174,7 +174,9 @@ func (an *AutoNAT) Close() {
 	an.wg.Wait()
 	an.srv.Close()
 	an.cli.Close()
+	an.mx.Lock()
 	an.peers = nil
+	an.mx.Unlock()
 }
 
 // GetReachability makes a single dial request for checking reachability for requested addresses
@@ -195,20 +197,9 @@ func (an *AutoNAT) GetReachability(ctx context.Context, reqs []Request) (Result,
 	} else {
 		filteredReqs = reqs
 	}
-	an.mx.Lock()
-	now := time.Now()
-	var p peer.ID
-	for pr := range an.peers.Shuffled() {
-		if t := an.throttlePeer[pr]; t.After(now) {
-			continue
-		}
-		p = pr
-		an.throttlePeer[p] = time.Now().Add(an.throttlePeerDuration)
-		break
-	}
-	an.mx.Unlock()
-	if p == "" {
-		return Result{}, ErrNoPeers
+	p, err := an.pickServer()
+	if err != nil {
+		return Result{}, err
 	}
 	res, err := an.cli.GetReachability(ctx, p, filteredReqs)
 	if err != nil {
@@ -226,6 +217,28 @@ func (an *AutoNAT) GetReachability(ctx context.Context, reqs []Request) (Result,
 	return res, nil
 }
 
+// pickServer returns an autonatv2 server that isn't currently throttled, and throttles
+// it for throttlePeerDuration. It returns ErrNoPeers if there is no such server, or if
+// autonat has been closed.
+func (an *AutoNAT) pickServer() (peer.ID, error) {
+	an.mx.Lock()
+	defer an.mx.Unlock()
+
+	// nil after Close; host shutdown can have in-flight reachability checks
+	if an.peers == nil {
+		return "", ErrNoPeers
+	}
+	now := time.Now()
+	for p := range an.peers.Shuffled() {
+		if t := an.throttlePeer[p]; t.After(now) {
+			continue
+		}
+		an.throttlePeer[p] = time.Now().Add(an.throttlePeerDuration)
+		return p, nil
+	}
+	return "", ErrNoPeers
+}
+
 func (an *AutoNAT) updatePeer(p peer.ID) {
 	an.mx.Lock()
 	defer an.mx.Unlock()

p2p/protocol/autonatv2/autonat_test.go

@@ -97,6 +97,16 @@ func TestAutoNATPrivateAddr(t *testing.T) {
 	require.ErrorIs(t, err, ErrPrivateAddrs)
 }
 
+func TestGetReachabilityAfterClose(t *testing.T) {
+	// The host closes autonat before the address manager, whose reachability
+	// tracker workers may still issue checks during shutdown.
+	an := newAutoNAT(t, nil)
+	an.Close()
+	res, err := an.GetReachability(context.Background(), []Request{{Addr: ma.StringCast("/ip4/1.2.3.4/udp/10/quic-v1")}})
+	require.ErrorIs(t, err, ErrNoPeers)
+	require.Equal(t, Result{}, res)
+}
+
 func TestClientRequest(t *testing.T) {
 	an := newAutoNAT(t, nil, AllowPrivateAddrs)
 	defer an.Close()