basichost: move observed address manager to basichost (#3332)

Author: sukunrt

config/config.go

@@ -28,6 +28,7 @@ import (
 	bhost "github.com/libp2p/go-libp2p/p2p/host/basic"
 	blankhost "github.com/libp2p/go-libp2p/p2p/host/blank"
 	"github.com/libp2p/go-libp2p/p2p/host/eventbus"
+	"github.com/libp2p/go-libp2p/p2p/host/observedaddrs"
 	"github.com/libp2p/go-libp2p/p2p/host/peerstore/pstoremem"
 	rcmgr "github.com/libp2p/go-libp2p/p2p/host/resource-manager"
 	routed "github.com/libp2p/go-libp2p/p2p/host/routed"
@@ -434,23 +435,23 @@ func (cfg *Config) addTransports() ([]fx.Option, error) {
 	return fxopts, nil
 }
 
-func (cfg *Config) newBasicHost(swrm *swarm.Swarm, eventBus event.Bus, an *autonatv2.AutoNAT) (*bhost.BasicHost, error) {
+func (cfg *Config) newBasicHost(swrm *swarm.Swarm, eventBus event.Bus, an *autonatv2.AutoNAT, o bhost.ObservedAddrsManager) (*bhost.BasicHost, error) {
 	h, err := bhost.NewHost(swrm, &bhost.HostOpts{
-		EventBus:                        eventBus,
-		ConnManager:                     cfg.ConnManager,
-		AddrsFactory:                    cfg.AddrsFactory,
-		NATManager:                      cfg.NATManager,
-		EnablePing:                      !cfg.DisablePing,
-		UserAgent:                       cfg.UserAgent,
-		ProtocolVersion:                 cfg.ProtocolVersion,
-		EnableHolePunching:              cfg.EnableHolePunching,
-		HolePunchingOptions:             cfg.HolePunchingOptions,
-		EnableRelayService:              cfg.EnableRelayService,
-		RelayServiceOpts:                cfg.RelayServiceOpts,
-		EnableMetrics:                   !cfg.DisableMetrics,
-		PrometheusRegisterer:            cfg.PrometheusRegisterer,
-		DisableIdentifyAddressDiscovery: cfg.DisableIdentifyAddressDiscovery,
-		AutoNATv2:                       an,
+		EventBus:             eventBus,
+		ConnManager:          cfg.ConnManager,
+		AddrsFactory:         cfg.AddrsFactory,
+		NATManager:           cfg.NATManager,
+		EnablePing:           !cfg.DisablePing,
+		UserAgent:            cfg.UserAgent,
+		ProtocolVersion:      cfg.ProtocolVersion,
+		EnableHolePunching:   cfg.EnableHolePunching,
+		HolePunchingOptions:  cfg.HolePunchingOptions,
+		EnableRelayService:   cfg.EnableRelayService,
+		RelayServiceOpts:     cfg.RelayServiceOpts,
+		EnableMetrics:        !cfg.DisableMetrics,
+		PrometheusRegisterer: cfg.PrometheusRegisterer,
+		AutoNATv2:            an,
+		ObservedAddrsManager: o,
 	})
 	if err != nil {
 		return nil, err
@@ -529,6 +530,25 @@ func (cfg *Config) NewNode() (host.Host, error) {
 			})
 			return sw, nil
 		}),
+		fx.Provide(func(eventBus event.Bus, s *swarm.Swarm, lifecycle fx.Lifecycle) (bhost.ObservedAddrsManager, error) {
+			if cfg.DisableIdentifyAddressDiscovery {
+				return nil, nil
+			}
+			o, err := observedaddrs.NewManager(eventBus, s)
+			if err != nil {
+				return nil, err
+			}
+			lifecycle.Append(fx.Hook{
+				OnStart: func(context.Context) error {
+					o.Start(s)
+					return nil
+				},
+				OnStop: func(context.Context) error {
+					return o.Close()
+				},
+			})
+			return o, nil
+		}),
 		fx.Provide(func() (*autonatv2.AutoNAT, error) {
 			if !cfg.EnableAutoNATv2 {
 				return nil, nil

p2p/host/basic/addrs_manager.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io"
 	"net"
 	"slices"
 	"sync"
@@ -20,11 +21,15 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 )
 
-const maxObservedAddrsPerListenAddr = 5
+const maxObservedAddrsPerListenAddr = 3
 
-type observedAddrsManager interface {
-	OwnObservedAddrs() []ma.Multiaddr
-	ObservedAddrsFor(local ma.Multiaddr) []ma.Multiaddr
+// addrChangeTickrInterval is the interval to recompute host addrs.
+var addrChangeTickrInterval = 5 * time.Second
+
+// ObservedAddrsManager maps our local listen addrs to externally observed addrs.
+type ObservedAddrsManager interface {
+	Addrs(minObservers int) []ma.Multiaddr
+	AddrsFor(local ma.Multiaddr) []ma.Multiaddr
 }
 
 type hostAddrs struct {
@@ -42,7 +47,7 @@ type addrsManager struct {
 	addrsFactory             AddrsFactory
 	listenAddrs              func() []ma.Multiaddr
 	addCertHashes            func([]ma.Multiaddr) []ma.Multiaddr
-	observedAddrsManager     observedAddrsManager
+	observedAddrsManager     ObservedAddrsManager
 	interfaceAddrs           *interfaceAddrsCache
 	addrsReachabilityTracker *addrsReachabilityTracker
 
@@ -72,7 +77,7 @@ func newAddrsManager(
 	addrsFactory AddrsFactory,
 	listenAddrs func() []ma.Multiaddr,
 	addCertHashes func([]ma.Multiaddr) []ma.Multiaddr,
-	observedAddrsManager observedAddrsManager,
+	observedAddrsManager ObservedAddrsManager,
 	addrsUpdatedChan chan struct{},
 	client autonatv2Client,
 	enableMetrics bool,
@@ -107,7 +112,6 @@ func newAddrsManager(
 }
 
 func (a *addrsManager) Start() error {
-	// TODO: add Start method to NATMgr
 	if a.addrsReachabilityTracker != nil {
 		err := a.addrsReachabilityTracker.Start()
 		if err != nil {
@@ -135,8 +139,6 @@ func (a *addrsManager) Close() {
 }
 
 func (a *addrsManager) NetNotifee() network.Notifiee {
-	// Updating addrs in sync provides the nice property that
-	// host.Addrs() just after host.Network().Listen(x) will return x
 	return &network.NotifyBundle{
 		ListenF:      func(network.Network, ma.Multiaddr) { a.updateAddrsSync() },
 		ListenCloseF: func(network.Network, ma.Multiaddr) { a.updateAddrsSync() },
@@ -159,34 +161,27 @@ func (a *addrsManager) updateAddrsSync() {
 	}
 }
 
-func (a *addrsManager) startBackgroundWorker() error {
-	autoRelayAddrsSub, err := a.bus.Subscribe(new(event.EvtAutoRelayAddrsUpdated), eventbus.Name("addrs-manager"))
+func (a *addrsManager) startBackgroundWorker() (retErr error) {
+	autoRelayAddrsSub, err := a.bus.Subscribe(new(event.EvtAutoRelayAddrsUpdated), eventbus.Name("addrs-manager autorelay sub"))
 	if err != nil {
 		return fmt.Errorf("error subscribing to auto relay addrs: %s", err)
 	}
-
-	autonatReachabilitySub, err := a.bus.Subscribe(new(event.EvtLocalReachabilityChanged), eventbus.Name("addrs-manager"))
+	mc := multiCloser{autoRelayAddrsSub}
+	autonatReachabilitySub, err := a.bus.Subscribe(new(event.EvtLocalReachabilityChanged), eventbus.Name("addrs-manager autonatv1 sub"))
 	if err != nil {
-		err1 := autoRelayAddrsSub.Close()
-		if err1 != nil {
-			err1 = fmt.Errorf("error closign autorelaysub: %w", err1)
-		}
-		err = fmt.Errorf("error subscribing to autonat reachability: %s", err)
-		return errors.Join(err, err1)
+		return errors.Join(
+			fmt.Errorf("error subscribing to autonat reachability: %s", err),
+			mc.Close(),
+		)
 	}
+	mc = append(mc, autonatReachabilitySub)
 
 	emitter, err := a.bus.Emitter(new(event.EvtHostReachableAddrsChanged), eventbus.Stateful)
 	if err != nil {
-		err1 := autoRelayAddrsSub.Close()
-		if err1 != nil {
-			err1 = fmt.Errorf("error closing autorelaysub: %w", err1)
-		}
-		err2 := autonatReachabilitySub.Close()
-		if err2 != nil {
-			err2 = fmt.Errorf("error closing autonat reachability: %w", err2)
-		}
-		err = fmt.Errorf("error subscribing to autonat reachability: %s", err)
-		return errors.Join(err, err1, err2)
+		return errors.Join(
+			fmt.Errorf("error creating reachability subscriber: %s", err),
+			mc.Close(),
+		)
 	}
 
 	var relayAddrs []ma.Multiaddr
@@ -230,6 +225,10 @@ func (a *addrsManager) background(autoRelayAddrsSub, autonatReachabilitySub even
 		if err != nil {
 			log.Warnf("error closing autonat reachability sub: %s", err)
 		}
+		err = emitter.Close()
+		if err != nil {
+			log.Warnf("error closing host reachability emitter: %s", err)
+		}
 	}()
 
 	ticker := time.NewTicker(addrChangeTickrInterval)
@@ -371,7 +370,8 @@ func (a *addrsManager) HolePunchAddrs() []ma.Multiaddr {
 	// AllAddrs may ignore observed addresses in favour of NAT mappings.
 	// Use both for hole punching.
 	if a.observedAddrsManager != nil {
-		addrs = append(addrs, a.observedAddrsManager.OwnObservedAddrs()...)
+		// For holepunching, include all the best addresses we know even ones with only 1 observer.
+		addrs = append(addrs, a.observedAddrsManager.Addrs(1)...)
 	}
 	addrs = ma.Unique(addrs)
 	return slices.DeleteFunc(addrs, func(a ma.Multiaddr) bool { return !manet.IsPublicAddr(a) })
@@ -406,7 +406,12 @@ func (a *addrsManager) getLocalAddrs() []ma.Multiaddr {
 
 	finalAddrs := make([]ma.Multiaddr, 0, 8)
 	finalAddrs = a.appendPrimaryInterfaceAddrs(finalAddrs, listenAddrs)
-	finalAddrs = a.appendNATAddrs(finalAddrs, listenAddrs, a.interfaceAddrs.All())
+	if a.natManager != nil {
+		finalAddrs = a.appendNATAddrs(finalAddrs, listenAddrs)
+	}
+	if a.observedAddrsManager != nil {
+		finalAddrs = a.appendObservedAddrs(finalAddrs, listenAddrs, a.interfaceAddrs.All())
+	}
 
 	// Remove "/p2p-circuit" addresses from the list.
 	// The p2p-circuit listener reports its address as just /p2p-circuit. This is
@@ -444,65 +449,36 @@ func (a *addrsManager) appendPrimaryInterfaceAddrs(dst []ma.Multiaddr, listenAdd
 // appendNATAddrs appends the NAT-ed addrs for the listenAddrs. For unspecified listen addrs it appends the
 // public address for all the interfaces.
 // Inferring WebTransport from QUIC depends on the observed address manager.
-//
-// TODO: Merge the natmgr and identify.ObservedAddrManager in to one NatMapper module.
-func (a *addrsManager) appendNATAddrs(dst []ma.Multiaddr, listenAddrs []ma.Multiaddr, ifaceAddrs []ma.Multiaddr) []ma.Multiaddr {
-	var obsAddrs []ma.Multiaddr
+func (a *addrsManager) appendNATAddrs(dst []ma.Multiaddr, listenAddrs []ma.Multiaddr) []ma.Multiaddr {
 	for _, listenAddr := range listenAddrs {
-		var natAddr ma.Multiaddr
-		if a.natManager != nil {
-			natAddr = a.natManager.GetMapping(listenAddr)
-		}
-
-		// The order of the cases below is important.
-		switch {
-		case natAddr == nil: // no nat mapping
-			dst = a.appendObservedAddrs(dst, listenAddr, ifaceAddrs)
-		case manet.IsIPUnspecified(natAddr):
-			log.Infof("NAT device reported an unspecified IP as it's external address: %s", natAddr)
-			_, natRest := ma.SplitFirst(natAddr)
-			obsAddrs = a.appendObservedAddrs(obsAddrs[:0], listenAddr, ifaceAddrs)
-			for _, addr := range obsAddrs {
-				obsIP, _ := ma.SplitFirst(addr)
-				if obsIP != nil && manet.IsPublicAddr(obsIP.Multiaddr()) {
-					dst = append(dst, obsIP.Encapsulate(natRest))
-				}
-			}
-		// This is !Public as opposed to IsPrivate intentionally.
-		// Public is a more restrictive classification in some cases, like IPv6 addresses which only
-		// consider unicast IPv6 addresses allocated so far as public(2000::/3).
-		case !manet.IsPublicAddr(natAddr): // nat reported non public addr(maybe CGNAT?)
-			// use both NAT and observed addr
-			dst = append(dst, natAddr)
-			dst = a.appendObservedAddrs(dst, listenAddr, ifaceAddrs)
-		default: // public addr
+		natAddr := a.natManager.GetMapping(listenAddr)
+		if natAddr != nil {
 			dst = append(dst, natAddr)
 		}
 	}
 	return dst
 }
 
-func (a *addrsManager) appendObservedAddrs(dst []ma.Multiaddr, listenAddr ma.Multiaddr, ifaceAddrs []ma.Multiaddr) []ma.Multiaddr {
-	if a.observedAddrsManager == nil {
-		return dst
-	}
-	// Add it for the listenAddr first.
+func (a *addrsManager) appendObservedAddrs(dst []ma.Multiaddr, listenAddrs, ifaceAddrs []ma.Multiaddr) []ma.Multiaddr {
+	// Add it for all the listenAddr first.
 	// listenAddr maybe unspecified. That's okay as connections on UDP transports
 	// will have the unspecified address as the local address.
-	obsAddrs := a.observedAddrsManager.ObservedAddrsFor(listenAddr)
-	if len(obsAddrs) > maxObservedAddrsPerListenAddr {
-		obsAddrs = obsAddrs[:maxObservedAddrsPerListenAddr]
+	for _, la := range listenAddrs {
+		obsAddrs := a.observedAddrsManager.AddrsFor(la)
+		if len(obsAddrs) > maxObservedAddrsPerListenAddr {
+			obsAddrs = obsAddrs[:maxObservedAddrsPerListenAddr]
+		}
+		dst = append(dst, obsAddrs...)
 	}
-	dst = append(dst, obsAddrs...)
 
 	// if it can be resolved into more addresses, add them too
-	resolved, err := manet.ResolveUnspecifiedAddress(listenAddr, ifaceAddrs)
+	resolved, err := manet.ResolveUnspecifiedAddresses(listenAddrs, ifaceAddrs)
 	if err != nil {
-		log.Warnf("failed to resolve listen addr %s, %s: %s", listenAddr, ifaceAddrs, err)
+		log.Warnf("failed to resolve listen addr %s, %s: %s", listenAddrs, ifaceAddrs, err)
 		return dst
 	}
 	for _, addr := range resolved {
-		obsAddrs = a.observedAddrsManager.ObservedAddrsFor(addr)
+		obsAddrs := a.observedAddrsManager.AddrsFor(addr)
 		if len(obsAddrs) > maxObservedAddrsPerListenAddr {
 			obsAddrs = obsAddrs[:maxObservedAddrsPerListenAddr]
 		}
@@ -678,3 +654,21 @@ func removeNotInSource(addrs, source []ma.Multiaddr) []ma.Multiaddr {
 	}
 	return addrs[:i]
 }
+
+type multiCloser []io.Closer
+
+func (mc *multiCloser) Close() error {
+	var errs []error
+	for _, closer := range *mc {
+		if err := closer.Close(); err != nil {
+			var closerName string
+			if named, ok := closer.(interface{ Name() string }); ok {
+				closerName = named.Name()
+			} else {
+				closerName = fmt.Sprintf("%T", closer)
+			}
+			errs = append(errs, fmt.Errorf("error closing %s: %w", closerName, err))
+		}
+	}
+	return errors.Join(errs...)
+}