Autopeering improvements

components/dashboard/component.go

@@ -17,7 +17,7 @@ import (
 	"github.com/iotaledger/hive.go/ierrors"
 	"github.com/iotaledger/iota-core/components/metricstracker"
 	"github.com/iotaledger/iota-core/pkg/daemon"
-	"github.com/iotaledger/iota-core/pkg/network/p2p"
+	"github.com/iotaledger/iota-core/pkg/network"
 	"github.com/iotaledger/iota-core/pkg/protocol"
 )
 
@@ -49,7 +49,7 @@ type dependencies struct {
 	Host           host.Host
 	Protocol       *protocol.Protocol
 	AppInfo        *app.Info
-	P2PManager     *p2p.Manager
+	NetworkManager network.Manager
 	MetricsTracker *metricstracker.MetricsTracker
 }
 
@@ -171,28 +171,20 @@ func currentNodeStatus() *nodestatus {
 
 func neighborMetrics() []neighbormetric {
 	var stats []neighbormetric
-	if deps.P2PManager == nil {
+	if deps.NetworkManager == nil {
 		return stats
 	}
 
 	// gossip plugin might be disabled
-	neighbors := deps.P2PManager.AllNeighbors()
+	neighbors := deps.NetworkManager.AllNeighbors()
 	if neighbors == nil {
 		return stats
 	}
 
 	for _, neighbor := range neighbors {
-		// origin := "Inbound"
-		// for _, p := range deps.P2PManager.AllNeighbors() {
-		//	if neighbor.Peer == peer {
-		//		origin = "Outbound"
-		//		break
-		//	}
-		// }
-
 		stats = append(stats, neighbormetric{
-			ID:             neighbor.Peer.ID.String(),
-			Addresses:      fmt.Sprintf("%s", neighbor.Peer.PeerAddresses),
+			ID:             neighbor.Peer().ID.String(),
+			Addresses:      fmt.Sprintf("%s", neighbor.Peer().PeerAddresses),
 			PacketsRead:    neighbor.PacketsRead(),
 			PacketsWritten: neighbor.PacketsWritten(),
 		})

components/p2p/component.go

@@ -2,9 +2,7 @@ package p2p
 
 import (
 	"context"
-	"fmt"
 	"path/filepath"
-	"time"
 
 	"github.com/libp2p/go-libp2p"
 	"github.com/libp2p/go-libp2p/core/crypto"
@@ -21,11 +19,8 @@ import (
 	"github.com/iotaledger/hive.go/ierrors"
 	"github.com/iotaledger/hive.go/kvstore"
 	hivedb "github.com/iotaledger/hive.go/kvstore/database"
-	"github.com/iotaledger/hive.go/runtime/event"
 	"github.com/iotaledger/iota-core/pkg/daemon"
 	"github.com/iotaledger/iota-core/pkg/network"
-	"github.com/iotaledger/iota-core/pkg/network/autopeering"
-	"github.com/iotaledger/iota-core/pkg/network/manualpeering"
 	"github.com/iotaledger/iota-core/pkg/network/p2p"
 	"github.com/iotaledger/iota-core/pkg/protocol"
 )
@@ -51,9 +46,7 @@ type dependencies struct {
 	dig.In
 	PeeringConfig        *configuration.Configuration `name:"peeringConfig"`
 	PeeringConfigManager *p2p.ConfigManager
-	ManualPeeringMgr     *manualpeering.Manager
-	AutoPeeringMgr       *autopeering.Manager
-	P2PManager           *p2p.Manager
+	NetworkManager       network.Manager
 	PeerDB               *network.DB
 	Protocol             *protocol.Protocol
 	PeerDBKVSTore        kvstore.KVStore `name:"peerDBKVStore"`
@@ -79,49 +72,6 @@ func initConfigParams(c *dig.Container) error {
 }
 
 func provide(c *dig.Container) error {
-	type manualPeeringDeps struct {
-		dig.In
-
-		P2PManager *p2p.Manager
-	}
-
-	if err := c.Provide(func(deps manualPeeringDeps) *manualpeering.Manager {
-		return manualpeering.NewManager(deps.P2PManager, Component.WorkerPool, Component.Logger)
-	}); err != nil {
-		return err
-	}
-
-	type autoPeeringDeps struct {
-		dig.In
-
-		Protocol   *protocol.Protocol
-		P2PManager *p2p.Manager
-		Host       host.Host
-		PeerDB     *network.DB
-	}
-
-	if err := c.Provide(func(deps autoPeeringDeps) *autopeering.Manager {
-		peersMultiAddresses, err := getMultiAddrsFromString(ParamsPeers.BootstrapPeers)
-		if err != nil {
-			Component.LogFatalf("Failed to parse bootstrapPeers param: %s", err)
-		}
-
-		for _, multiAddr := range peersMultiAddresses {
-			bootstrapPeer, err := network.NewPeerFromMultiAddr(multiAddr)
-			if err != nil {
-				Component.LogFatalf("Failed to parse bootstrap peer multiaddress: %s", err)
-			}
-
-			if err := deps.PeerDB.UpdatePeer(bootstrapPeer); err != nil {
-				Component.LogErrorf("Failed to update bootstrap peer: %s", err)
-			}
-		}
-
-		return autopeering.NewManager(deps.Protocol.LatestAPI().ProtocolParameters().NetworkName(), deps.P2PManager, deps.Host, deps.PeerDB, Component.Logger)
-	}); err != nil {
-		return err
-	}
-
 	type peerDatabaseResult struct {
 		dig.Out
 
@@ -251,7 +201,7 @@ func provide(c *dig.Container) error {
 		connManager, err := connmgr.NewConnManager(
 			ParamsP2P.ConnectionManager.LowWatermark,
 			ParamsP2P.ConnectionManager.HighWatermark,
-			connmgr.WithGracePeriod(time.Minute),
+			connmgr.WithEmergencyTrim(true),
 		)
 		if err != nil {
 			Component.LogPanicf("unable to initialize connection manager: %s", err)
@@ -263,6 +213,7 @@ func provide(c *dig.Container) error {
 			libp2p.Transport(tcp.NewTCPTransport),
 			libp2p.ConnectionManager(connManager),
 			libp2p.NATPortMap(),
+			libp2p.DisableRelay(),
 			// Define a custom address factory to inject external addresses to the DHT advertisements.
 			libp2p.AddrsFactory(func() func(addrs []multiaddr.Multiaddr) []multiaddr.Multiaddr {
 				var externalMultiAddrs []multiaddr.Multiaddr
@@ -294,8 +245,31 @@ func provide(c *dig.Container) error {
 		Component.LogPanic(err.Error())
 	}
 
-	return c.Provide(func(host host.Host, peerDB *network.DB) *p2p.Manager {
-		return p2p.NewManager(host, peerDB, Component.Logger)
+	type p2pManagerDeps struct {
+		dig.In
+		Host   host.Host
+		PeerDB *network.DB
+	}
+
+	return c.Provide(func(inDeps p2pManagerDeps) network.Manager {
+
+		peersMultiAddresses, err := getMultiAddrsFromString(ParamsPeers.BootstrapPeers)
+		if err != nil {
+			Component.LogFatalf("Failed to parse bootstrapPeers param: %s", err)
+		}
+
+		for _, multiAddr := range peersMultiAddresses {
+			bootstrapPeer, err := network.NewPeerFromMultiAddr(multiAddr)
+			if err != nil {
+				Component.LogFatalf("Failed to parse bootstrap peer multiaddress: %s", err)
+			}
+
+			if err := inDeps.PeerDB.UpdatePeer(bootstrapPeer); err != nil {
+				Component.LogErrorf("Failed to update bootstrap peer: %s", err)
+			}
+		}
+
+		return p2p.NewManager(inDeps.Host, inDeps.PeerDB, ParamsP2P.Autopeering.MaxPeers, Component.Logger)
 	})
 }
 
@@ -321,53 +295,27 @@ func configure() error {
 	}
 
 	// log the p2p events
-	deps.P2PManager.Events.NeighborAdded.Hook(func(neighbor *p2p.Neighbor) {
-		Component.LogInfof("Neighbor added: %s / %s", neighbor.PeerAddresses, neighbor.ID)
-	}, event.WithWorkerPool(Component.WorkerPool))
+	deps.NetworkManager.OnNeighborAdded(func(neighbor network.Neighbor) {
+		Component.LogInfof("neighbor added: %s / %s", neighbor.Peer().PeerAddresses, neighbor.Peer().ID)
+	})
 
-	deps.P2PManager.Events.NeighborRemoved.Hook(func(neighbor *p2p.Neighbor) {
-		Component.LogInfof("Neighbor removed: %s / %s", neighbor.PeerAddresses, neighbor.ID)
-	}, event.WithWorkerPool(Component.WorkerPool))
+	deps.NetworkManager.OnNeighborRemoved(func(neighbor network.Neighbor) {
+		Component.LogInfof("neighbor removed: %s / %s", neighbor.Peer().PeerAddresses, neighbor.Peer().ID)
+	})
 
 	return nil
 }
 
 func run() error {
 	if err := Component.Daemon().BackgroundWorker(Component.Name, func(ctx context.Context) {
-		deps.ManualPeeringMgr.Start()
-
-		defer func() {
-			if err := deps.ManualPeeringMgr.Stop(); err != nil {
-				Component.LogErrorf("Failed to stop the manager: %s", err)
-			}
-		}()
+		defer deps.NetworkManager.Shutdown()
 
-		if ParamsP2P.Autopeering.Enabled {
-			if err := deps.AutoPeeringMgr.Start(ctx); err != nil {
-				Component.LogFatalf("Failed to start autopeering manager: %s", err)
-			}
-
-			defer func() {
-				if err := deps.AutoPeeringMgr.Stop(); err != nil {
-					Component.LogErrorf("Failed to stop autopeering manager: %s", err)
-				}
-			}()
+		if err := deps.NetworkManager.Start(ctx, deps.Protocol.LatestAPI().ProtocolParameters().NetworkName()); err != nil {
+			Component.LogFatalf("Failed to start p2p manager: %s", err)
 		}
 
 		//nolint:contextcheck // false positive
 		connectConfigKnownPeers()
-		<-ctx.Done()
-	}, daemon.PriorityManualPeering); err != nil {
-		Component.LogFatalf("Failed to start as daemon: %s", err)
-	}
-
-	if err := Component.Daemon().BackgroundWorker(fmt.Sprintf("%s-P2PManager", Component.Name), func(ctx context.Context) {
-		defer deps.P2PManager.Shutdown()
-		defer func() {
-			if err := deps.P2PManager.P2PHost().Close(); err != nil {
-				Component.LogWarnf("Failed to close libp2p host: %+v", err)
-			}
-		}()
 
 		<-ctx.Done()
 	}, daemon.PriorityP2P); err != nil {
@@ -405,7 +353,7 @@ func connectConfigKnownPeers() {
 			Component.LogPanicf("invalid peer address info: %s", err)
 		}
 
-		if err := deps.ManualPeeringMgr.AddPeers(multiAddr); err != nil {
+		if err := deps.NetworkManager.AddManualPeers(multiAddr); err != nil {
 			Component.LogInfof("failed to add peer: %s, error: %s", multiAddr.String(), err)
 		}
 	}

components/p2p/params.go

@@ -29,7 +29,7 @@ type ParametersP2P struct {
 	IdentityPrivateKey string `default:"" usage:"private key used to derive the node identity (optional)"`
 
 	Autopeering struct {
-		Enabled bool `default:"true" usage:"enable or disable autopeering"`
+		MaxPeers int `default:"5" usage:"the max number of autopeer connections. Set to 0 to disable autopeering."`
 	}
 
 	Database struct {

components/protocol/component.go

@@ -14,7 +14,7 @@ import (
 	"github.com/iotaledger/hive.go/runtime/workerpool"
 	"github.com/iotaledger/iota-core/pkg/daemon"
 	"github.com/iotaledger/iota-core/pkg/model"
-	"github.com/iotaledger/iota-core/pkg/network/p2p"
+	"github.com/iotaledger/iota-core/pkg/network"
 	"github.com/iotaledger/iota-core/pkg/protocol"
 	"github.com/iotaledger/iota-core/pkg/protocol/engine/attestation/slotattestation"
 	"github.com/iotaledger/iota-core/pkg/protocol/engine/filter/presolidfilter/presolidblockfilter"
@@ -115,7 +115,7 @@ func provide(c *dig.Container) error {
 
 		DatabaseEngine     hivedb.Engine `name:"databaseEngine"`
 		ProtocolParameters []iotago.ProtocolParameters
-		P2PManager         *p2p.Manager
+		NetworkManager     network.Manager
 	}
 
 	return c.Provide(func(deps protocolDeps) *protocol.Protocol {
@@ -132,7 +132,7 @@ func provide(c *dig.Container) error {
 		return protocol.New(
 			Component.Logger,
 			workerpool.NewGroup("Protocol"),
-			deps.P2PManager,
+			deps.NetworkManager,
 			protocol.WithBaseDirectory(ParamsDatabase.Path),
 			protocol.WithStorageOptions(
 				storage.WithDBEngine(deps.DatabaseEngine),

go.mod

@@ -144,7 +144,7 @@ require (
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/pokt-network/smt v0.9.2 // indirect
 	github.com/polydawn/refmt v0.89.0 // indirect
-	github.com/prometheus/client_model v0.5.0 // indirect
+	github.com/prometheus/client_model v0.6.0 // indirect
 	github.com/prometheus/common v0.47.0 // indirect
 	github.com/prometheus/procfs v0.12.0 // indirect
 	github.com/quic-go/qpack v0.4.0 // indirect

go.sum

@@ -547,8 +547,8 @@ github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:
 github.com/prometheus/client_model v0.0.0-20190129233127-fd36f4220a90/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
 github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
 github.com/prometheus/client_model v0.2.0/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
-github.com/prometheus/client_model v0.5.0 h1:VQw1hfvPvk3Uv6Qf29VrPF32JB6rtbgI6cYPYQjL0Qw=
-github.com/prometheus/client_model v0.5.0/go.mod h1:dTiFglRmd66nLR9Pv9f0mZi7B7fk5Pm3gvsjB5tr+kI=
+github.com/prometheus/client_model v0.6.0 h1:k1v3CzpSRUTrKMppY35TLwPvxHqBu0bYgxZzqGIgaos=
+github.com/prometheus/client_model v0.6.0/go.mod h1:NTQHnmxFpouOD0DpvP4XujX3CdOAGQPoaGhyTchlyt8=
 github.com/prometheus/common v0.0.0-20180801064454-c7de2306084e/go.mod h1:daVV7qP5qjZbuso7PdcryaAu0sAZbrN9i7WWcTMWvro=
 github.com/prometheus/common v0.4.1/go.mod h1:TNfzLD0ON7rHzMJeJkieUDPYmFC7Snx/y86RQel1bk4=
 github.com/prometheus/common v0.10.0/go.mod h1:Tlit/dnDKsSWFlCLTWaA1cyBgKHSMdTB80sz/V91rCo=

pkg/daemon/shutdown.go

@@ -5,12 +5,8 @@ package daemon
 
 const (
 	PriorityCloseDatabase = iota // no dependencies
-	PriorityPeerDatabase
 	PriorityP2P
-	PriorityManualPeering
 	PriorityProtocol
-	PriorityBlockIssuer
-	PriorityActivity // depends on BlockIssuer
 	PriorityRestAPI
 	PriorityINX
 	PriorityDashboardMetrics

pkg/network/errors.go

@@ -0,0 +1,16 @@
+package network
+
+import "github.com/iotaledger/hive.go/ierrors"
+
+var (
+	// ErrNotRunning is returned when a peer is added to a stopped or not yet started network manager.
+	ErrNotRunning = ierrors.New("manager not running")
+	// ErrUnknownPeer is returned when the specified peer is not known to the network manager.
+	ErrUnknownPeer = ierrors.New("unknown neighbor")
+	// ErrLoopbackPeer is returned when the own peer is added.
+	ErrLoopbackPeer = ierrors.New("loopback connection not allowed")
+	// ErrDuplicatePeer is returned when the same peer is added more than once.
+	ErrDuplicatePeer = ierrors.New("already connected")
+	// ErrMaxAutopeeringPeersReached is returned when the maximum number of autopeering peers is reached.
+	ErrMaxAutopeeringPeersReached = ierrors.New("max autopeering peers reached")
+)

pkg/network/manager.go

@@ -0,0 +1,33 @@
+package network
+
+import (
+	"context"
+
+	"github.com/libp2p/go-libp2p/core/host"
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/multiformats/go-multiaddr"
+
+	"github.com/iotaledger/hive.go/runtime/event"
+)
+
+type Manager interface {
+	Endpoint
+
+	DialPeer(context.Context, *Peer) error
+
+	OnNeighborAdded(func(Neighbor)) *event.Hook[func(Neighbor)]
+	OnNeighborRemoved(func(Neighbor)) *event.Hook[func(Neighbor)]
+
+	AllNeighbors() []Neighbor
+	AutopeeringNeighborsCount() int
+
+	DropNeighbor(peer.ID) error
+	NeighborExists(peer.ID) bool
+
+	P2PHost() host.Host
+
+	Start(ctx context.Context, networkID string) error
+	Shutdown()
+
+	AddManualPeers(...multiaddr.Multiaddr) error
+}

pkg/network/neighbor.go

@@ -0,0 +1,7 @@
+package network
+
+type Neighbor interface {
+	Peer() *Peer
+	PacketsRead() uint64
+	PacketsWritten() uint64
+}