feat(SPV-742): add peer manager

cmd/main.go

@@ -42,7 +42,7 @@ var version = "development"
 
 type P2PServer interface {
 	Start() error
-	Shutdown() error
+	Shutdown()
 }
 
 // nolint: godot
@@ -172,9 +172,7 @@ func main() {
 
 	<-quit
 
-	if err := p2pServer.Shutdown(); err != nil {
-		log.Error().Msgf("failed to stop p2p server: %v", err)
-	}
+	p2pServer.Shutdown()
 
 	if err := ws.Shutdown(); err != nil {
 		log.Error().Msgf("failed to stop websocket server: %v", err)

config/config.go

@@ -126,6 +126,10 @@ 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"`
+	MaxInboundConnections  uint `mapstructure:"max_inbound_connections" description:"Maximum active inbound connections"`
+	AcceptLocalPeers       bool `mapstructure:"accept_local_peers" description:"Accept connection from local network"`
 }
 
 // LoggingConfig represents a logging config.

config/defaults.go

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

internal/chaincfg/params.go

@@ -108,7 +108,7 @@ type Params struct {
 	Net wire.BitcoinNet
 
 	// DefaultPort defines the default peer-to-peer port for the network.
-	DefaultPort string
+	DefaultPort uint16
 
 	// DNSSeeds defines a list of DNS seeds for the network that are used
 	// as one method to discover peers.
@@ -222,7 +222,7 @@ type Params struct {
 var MainNetParams = Params{
 	Name:        "mainnet",
 	Net:         wire.MainNet,
-	DefaultPort: "8333",
+	DefaultPort: uint16(8333),
 	DNSSeeds: []DNSSeed{
 		{"seed-nodes.bsvb.tech", true},
 	},
@@ -297,7 +297,7 @@ var MainNetParams = Params{
 var RegressionNetParams = Params{
 	Name:        "regtest",
 	Net:         wire.TestNet,
-	DefaultPort: "18444",
+	DefaultPort: uint16(18444),
 	DNSSeeds:    []DNSSeed{},
 
 	// Chain parameters
@@ -365,7 +365,7 @@ var RegressionNetParams = Params{
 var TestNet3Params = Params{
 	Name:        "testnet3",
 	Net:         wire.TestNet3,
-	DefaultPort: "18333",
+	DefaultPort: uint16(18333),
 	DNSSeeds: []DNSSeed{
 		{"testnet-seed.metasv.io", true},
 		{"testnet-btccash-seeder.bitcoinunlimited.info", true},
@@ -452,7 +452,7 @@ var TestNet3Params = Params{
 var SimNetParams = Params{
 	Name:        "simnet",
 	Net:         wire.SimNet,
-	DefaultPort: "18555",
+	DefaultPort: uint16(18555),
 	DNSSeeds:    []DNSSeed{}, // NOTE: There must NOT be any seeds.
 
 	// Chain parameters

internal/transports/p2p/network/address_book.go

@@ -0,0 +1,165 @@
+package network
+
+import (
+	"fmt"
+	"math/rand"
+	"sync"
+	"time"
+
+	"github.com/bitcoin-sv/block-headers-service/internal/wire"
+)
+
+type addressBucketType string
+
+const (
+	freeBucket   addressBucketType = "free"
+	usedBucket   addressBucketType = "used"
+	bannedBucket addressBucketType = "banned"
+)
+
+// AddressBook represents a collection of known network addresses.
+type AddressBook struct {
+	banDuration  time.Duration
+	addrs        map[addressBucketType]*addrBucket
+	mu           sync.Mutex
+	addrFitlerFn func(*wire.NetAddress) bool
+}
+
+// NewAddressBook creates and initializes a new AddressBook instance.
+func NewAddressBook(banDuration time.Duration, acceptLocalAddresses bool) *AddressBook {
+	// Set the address filter function based on whether local addresses are accepted
+	addrFilterFn := wire.IsRoutable
+	if acceptLocalAddresses {
+		addrFilterFn = wire.IsRoutableWithLocal
+	}
+
+	const addressesInitCapacity = 500
+	const usedAddressesInitCapacity = 8
+
+	knownAddress := make(map[addressBucketType]*addrBucket, 3)
+	knownAddress[freeBucket] = newAddrBucket(addressesInitCapacity)
+	knownAddress[bannedBucket] = newAddrBucket(addressesInitCapacity)
+	knownAddress[usedBucket] = newAddrBucket(usedAddressesInitCapacity)
+
+	return &AddressBook{
+		banDuration:  banDuration,
+		addrFitlerFn: addrFilterFn,
+		addrs:        knownAddress,
+	}
+}
+
+// UpsertAddrs updates or adds multiple addresses.
+func (a *AddressBook) UpsertAddrs(address []*wire.NetAddress) {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+
+	for _, addr := range address {
+		if !a.addrFitlerFn(addr) {
+			continue
+		}
+
+		key, ka, _ := a.findAddr(addr)
+		// If the address is not found, add it to the AddressBook.
+		if ka == nil {
+			a.addrs[freeBucket].add(key, &knownAddress{addr: addr})
+		} else if addr.Timestamp.After(ka.addr.Timestamp) {
+			// Otherwise, update the timestamp if the new one is newer.
+			ka.addr.Timestamp = addr.Timestamp
+		}
+	}
+}
+
+// MarkUsedAddr updates or adds a peer's address.
+func (a *AddressBook) MarkUsedAddr(pa *wire.NetAddress) {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+
+	key := addrKey(pa)
+	// remove from free if exists
+	a.addrs[freeBucket].rm(key)
+	// add to used
+	a.addrs[usedBucket].add(key, &knownAddress{addr: pa})
+
+}
+
+// BanAddr bans a network address. Ignores address if doesn't exist in the AddressBook.
+func (a *AddressBook) BanAddr(addr *wire.NetAddress) {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+
+	if key, ka, bucket := a.findAddr(addr); ka != nil {
+		switch bucket {
+		case freeBucket:
+			a.ban(bucket, key, ka)
+		case usedBucket:
+			a.ban(bucket, key, ka)
+		case bannedBucket:
+		default:
+			// Do nothing
+		}
+	}
+}
+
+// GetRandFreeAddr returns a randomly chosen unused network address.
+func (a *AddressBook) GetRandFreeAddr() *wire.NetAddress {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+
+	freeAddres := a.addrs[freeBucket].items
+	fLen := len(freeAddres)
+	if fLen == 0 {
+		return nil
+	}
+
+	// #nosec G404
+	randIndx := rand.Intn(fLen)
+	return freeAddres[randIndx].addr
+}
+
+func (a *AddressBook) findAddr(addr *wire.NetAddress) (key string, ka *knownAddress, bucket addressBucketType) {
+	key = addrKey(addr)
+
+	// search in free addresses
+	if ka = a.addrs[freeBucket].find(key); ka != nil {
+		bucket = freeBucket
+		return
+	}
+
+	// search in used
+	if ka = a.addrs[usedBucket].find(key); ka != nil {
+		bucket = usedBucket
+		return
+	}
+
+	// search in banned
+	if ka = a.addrs[bannedBucket].find(key); ka != nil {
+		bucket = bannedBucket
+		return
+	}
+
+	return key, nil, ""
+}
+
+func (a *AddressBook) ban(bucket addressBucketType, key string, ka *knownAddress) {
+	a.addrs[bucket].rm(key)
+	a.addrs[bannedBucket].add(key, ka)
+	go a.unban(key, ka)
+}
+
+func (a *AddressBook) unban(key string, ka *knownAddress) {
+	time.Sleep(a.banDuration)
+
+	a.mu.Lock()
+	defer a.mu.Unlock()
+
+	a.addrs[bannedBucket].rm(key)
+	a.addrs[freeBucket].add(key, ka)
+}
+
+func addrKey(addr *wire.NetAddress) string {
+	return fmt.Sprintf("%s:%d", addr.IP, addr.Port)
+}
+
+type knownAddress struct {
+	addr *wire.NetAddress
+}