diff --git a/p2p/host/basic/addrs_manager.go b/p2p/host/basic/addrs_manager.go index cc90de4012..3abeec22c2 100644 --- a/p2p/host/basic/addrs_manager.go +++ b/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) } diff --git a/p2p/host/basic/addrs_reachability_tracker.go b/p2p/host/basic/addrs_reachability_tracker.go index 5e68847df4..a98ed30a20 100644 --- a/p2p/host/basic/addrs_reachability_tracker.go +++ b/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 diff --git a/p2p/host/basic/addrs_reachability_tracker_test.go b/p2p/host/basic/addrs_reachability_tracker_test.go index 5e88e6c489..c3af935063 100644 --- a/p2p/host/basic/addrs_reachability_tracker_test.go +++ b/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) { diff --git a/p2p/protocol/autonatv2/autonat.go b/p2p/protocol/autonatv2/autonat.go index 95f78c9329..9aa04d45f6 100644 --- a/p2p/protocol/autonatv2/autonat.go +++ b/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() diff --git a/p2p/protocol/autonatv2/autonat_test.go b/p2p/protocol/autonatv2/autonat_test.go index dca55d2140..4bccb25404 100644 --- a/p2p/protocol/autonatv2/autonat_test.go +++ b/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()