feat(SPV-742): add peer manager

config/config.go

@@ -126,6 +126,8 @@ type P2PConfig struct {
 	UserAgentName         string        `mapstructure:"user_agent_name" description:"The name that should be used during announcement of the client on the p2p network"`
 	UserAgentVersion      string        `mapstructure:"user_agent_version" description:"By default will be equal to application version, but can be overridden for development purposes"`
 	Experimental          bool          `mapstructure:"experimental" description:"Turns on a new (highly experimental) way of getting headers with the usage of /internal/transports/p2p instead of /transports/p2p"`
+
+	MaxOutboundConnections uint `mapstructure:"max_outbound_connections" description:"Maximum active outbound connections"`
 }
 
 // LoggingConfig represents a logging config.

config/defaults.go

@@ -72,6 +72,7 @@ func getP2PDefaults() *P2PConfig {
 		UserAgentName:             ApplicationName,
 		UserAgentVersion:          Version(),
 		Experimental:              false,
+		MaxOutboundConnections:    8,
 	}
 }
 

internal/transports/p2p/network/address_book.go

@@ -0,0 +1,120 @@
+package network
+
+import (
+	"fmt"
+	"math/rand"
+	"sync"
+	"time"
+
+	"github.com/bitcoin-sv/block-headers-service/internal/transports/p2p/peer"
+	"github.com/bitcoin-sv/block-headers-service/internal/wire"
+)
+
+type AddressBook struct {
+	banDuration time.Duration
+	addrs       []*knownAddress
+	keyIndex    map[string]int
+
+	mu sync.Mutex
+}
+
+func NewAdressbook(banDuration time.Duration) *AddressBook {
+	return &AddressBook{
+		addrs:       make([]*knownAddress, 0, 400),
+		keyIndex:    make(map[string]int, 400),
+		banDuration: banDuration,
+	}
+}
+
+func (a *AddressBook) UpsertPeerAddr(p *peer.Peer) {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+
+	pa := p.GetPeerAddr()
+	key, ka := a.internalFind(pa)
+
+	if ka != nil {
+		ka.peer = p
+	} else {
+		a.internalAddAddr(key, &knownAddress{addr: pa, peer: p})
+	}
+}
+
+func (a *AddressBook) AddAddrs(address []*wire.NetAddress) {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+
+	for _, addr := range address {
+		if !IsRoutable(addr) {
+			continue
+		}
+
+		key, ka := a.internalFind(addr)
+		if ka == nil {
+			a.internalAddAddr(key, &knownAddress{addr: addr})
+		} else if addr.Timestamp.After(ka.addr.Timestamp) {
+			ka.addr.Timestamp = addr.Timestamp
+		}
+	}
+}
+
+func (a *AddressBook) BanAddr(addr *wire.NetAddress) {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+
+	_, ka := a.internalFind(addr)
+	if ka != nil {
+		now := time.Now()
+		ka.banTimestamp = &now
+	}
+}
+
+func (a *AddressBook) GetRndUnusedAddr(tries uint) *wire.NetAddress {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+
+	alen := len(a.addrs)
+	for i := uint(0); i < tries; i++ {
+		ka := a.addrs[rand.Intn(alen)]
+		if ka.peer == nil {
+			if ka.isBanned(a.banDuration) {
+				continue
+			}
+			return ka.addr
+		}
+	}
+
+	return nil
+}
+
+func (a *AddressBook) internalFind(addr *wire.NetAddress) (string, *knownAddress) {
+	key := addrKey(addr)
+	addrIndex, ok := a.keyIndex[key]
+
+	if ok {
+		return key, a.addrs[addrIndex]
+	}
+	return key, nil
+}
+
+func (a *AddressBook) internalAddAddr(key string, addr *knownAddress) {
+	newItemIndex := len(a.addrs)
+
+	a.addrs = append(a.addrs, addr)
+	a.keyIndex[key] = newItemIndex
+}
+
+func addrKey(addr *wire.NetAddress) string {
+	return fmt.Sprintf("%s:%d", addr.IP, addr.Port)
+}
+
+type knownAddress struct {
+	addr *wire.NetAddress
+	peer *peer.Peer
+
+	banTimestamp *time.Time
+}
+
+func (a *knownAddress) isBanned(duration time.Duration) bool {
+	return a.banTimestamp != nil && time.Since(*a.banTimestamp) < duration
+}