func (k *Kad) pruneOversaturatedBins(depth uint8) {
	for i := range k.commonBinPrefixes {
		if i >= int(depth) {
			return
		}
		binPeersCount := k.connectedPeers.BinSize(uint8(i))
		if binPeersCount < overSaturationPeers {
			continue
		}
		binPeers := k.connectedPeers.BinPeers(uint8(i))
		peersToRemove := binPeersCount - overSaturationPeers
		for j := 0; peersToRemove > 0 && j < len(k.commonBinPrefixes[i]); j++ {
			pseudoAddr := k.commonBinPrefixes[i][j]
			peers := k.balancedSlotPeers(pseudoAddr, binPeers, i)
			if len(peers) <= 1 {
				continue
			}
			var smallestDuration time.Duration
			var newestPeer swarm.Address
			for _, peer := range peers {
				duration := k.collector.Inspect(peer).SessionConnectionDuration
				if smallestDuration == 0 || duration < smallestDuration {
					smallestDuration = duration
					newestPeer = peer
				}
			}
			err := k.p2p.Disconnect(newestPeer, "pruned from oversaturated bin")
			if err != nil {
				k.logger.Debugf("prune disconnect fail %v", err)
			}
			peersToRemove--
		}
	}
}
pruneOversaturatedBins disconnects out of depth peers from oversaturated bins
while maintaining the balance of the bin and favoring peers with longers connections
package kademlia
import (
	"context"
	random "crypto/rand"
	"encoding/json"
	"errors"
	"fmt"
	"math/big"
	"net"
	"sync"
	"syscall"
	"time"
	"github.com/ethersphere/bee/pkg/addressbook"
	"github.com/ethersphere/bee/pkg/blocker"
	"github.com/ethersphere/bee/pkg/discovery"
	"github.com/ethersphere/bee/pkg/logging"
	"github.com/ethersphere/bee/pkg/p2p"
	"github.com/ethersphere/bee/pkg/pingpong"
	"github.com/ethersphere/bee/pkg/shed"
	"github.com/ethersphere/bee/pkg/swarm"
	"github.com/ethersphere/bee/pkg/topology"
	im "github.com/ethersphere/bee/pkg/topology/kademlia/internal/metrics"
	"github.com/ethersphere/bee/pkg/topology/kademlia/internal/waitnext"
	"github.com/ethersphere/bee/pkg/topology/pslice"
	ma "github.com/multiformats/go-multiaddr"
	"golang.org/x/sync/errgroup"
)
const (
	maxConnAttempts        = 1  
	maxBootNodeAttempts    = 3  
	defaultBitSuffixLength = 3  
	addPeerBatchSize = 500
	peerConnectionAttemptTimeout = 5 * time.Second  
	flagTimeout      = 5 * time.Minute   
	blockDuration    = time.Hour         
	blockWorkerWakup = time.Second * 10  
)
var (
	nnLowWatermark              = 2  
	quickSaturationPeers        = 4
	saturationPeers             = 8
	overSaturationPeers         = 20
	bootNodeOverSaturationPeers = 20
	shortRetry                  = 30 * time.Second
	timeToRetry                 = 2 * shortRetry
	broadcastBinSize            = 4
	peerPingPollTime            = 10 * time.Second  
)
var (
	errOverlayMismatch   = errors.New("overlay mismatch")
	errPruneEntry        = errors.New("prune entry")
	errEmptyBin          = errors.New("empty bin")
	errAnnounceLightNode = errors.New("announcing light node")
)
type (
	binSaturationFunc  func(bin uint8, peers, connected *pslice.PSlice) (saturated bool, oversaturated bool)
	sanctionedPeerFunc func(peer swarm.Address) bool
	pruneFunc          func(depth uint8)
	staticPeerFunc     func(peer swarm.Address) bool
)
var noopSanctionedPeerFn = func(_ swarm.Address) bool { return false }
type Options struct {
	SaturationFunc  binSaturationFunc
	Bootnodes       []ma.Multiaddr
	BootnodeMode    bool
	BitSuffixLength int
	PruneFunc       pruneFunc
	StaticNodes     []swarm.Address
}
type Kad struct {
	base              swarm.Address          
	discovery         discovery.Driver       
	addressBook       addressbook.Interface  
	p2p               p2p.Service            
	saturationFunc    binSaturationFunc      
	bitSuffixLength   int                    
	commonBinPrefixes [][]swarm.Address      
	connectedPeers    *pslice.PSlice         
	knownPeers        *pslice.PSlice         
	bootnodes         []ma.Multiaddr
	depth             uint8          
	radius            uint8          
	depthMu           sync.RWMutex   
	manageC           chan struct{}  
	peerSig           []chan struct{}
	peerSigMtx        sync.Mutex
	logger            logging.Logger  
	bootnode          bool            
	collector         *im.Collector
	quit              chan struct{}  
	halt              chan struct{}  
	done              chan struct{}  
	wg                sync.WaitGroup
	waitNext          *waitnext.WaitNext
	metrics           metrics
	pruneFunc         pruneFunc  
	pinger            pingpong.Interface
	staticPeer        staticPeerFunc
	bgBroadcastCtx    context.Context
	bgBroadcastCancel context.CancelFunc
	blocker           *blocker.Blocker
}
func New(
	base swarm.Address,
	addressbook addressbook.Interface,
	discovery discovery.Driver,
	p2p p2p.Service,
	pinger pingpong.Interface,
	metricsDB *shed.DB,
	logger logging.Logger,
	o Options,
) (*Kad, error) {
	if o.SaturationFunc == nil {
		os := overSaturationPeers
		if o.BootnodeMode {
			os = bootNodeOverSaturationPeers
		}
		o.SaturationFunc = binSaturated(os, isStaticPeer(o.StaticNodes))
	}
	if o.BitSuffixLength == 0 {
		o.BitSuffixLength = defaultBitSuffixLength
	}
	start := time.Now()
	imc, err := im.NewCollector(metricsDB)
	if err != nil {
		return nil, err
	}
	logger.Debugf("kademlia: NewCollector(...) took %v", time.Since(start))
	k := &Kad{
		base:              base,
		discovery:         discovery,
		addressBook:       addressbook,
		p2p:               p2p,
		saturationFunc:    o.SaturationFunc,
		bitSuffixLength:   o.BitSuffixLength,
		commonBinPrefixes: make([][]swarm.Address, int(swarm.MaxBins)),
		connectedPeers:    pslice.New(int(swarm.MaxBins), base),
		knownPeers:        pslice.New(int(swarm.MaxBins), base),
		bootnodes:         o.Bootnodes,
		manageC:           make(chan struct{}, 1),
		waitNext:          waitnext.New(),
		logger:            logger,
		bootnode:          o.BootnodeMode,
		collector:         imc,
		quit:              make(chan struct{}),
		halt:              make(chan struct{}),
		done:              make(chan struct{}),
		metrics:           newMetrics(),
		pruneFunc:         o.PruneFunc,
		pinger:            pinger,
		staticPeer:        isStaticPeer(o.StaticNodes),
	}
	blocklistCallback := func(a swarm.Address) {
		k.logger.Debugf("kademlia: disconnecting peer %s for ping failure", a.String())
		k.metrics.Blocklist.Inc()
	}
	k.blocker = blocker.New(p2p, flagTimeout, blockDuration, blockWorkerWakup, blocklistCallback, logger)
	if k.pruneFunc == nil {
		k.pruneFunc = k.pruneOversaturatedBins
	}
	if k.bitSuffixLength > 0 {
		k.commonBinPrefixes = generateCommonBinPrefixes(k.base, k.bitSuffixLength)
	}
	k.bgBroadcastCtx, k.bgBroadcastCancel = context.WithCancel(context.Background())
	return k, nil
}
type peerConnInfo struct {
	po   uint8
	addr swarm.Address
}
func (k *Kad) connectBalanced(wg *sync.WaitGroup, peerConnChan chan<- *peerConnInfo) {
	skipPeers := func(peer swarm.Address) bool {
		if k.waitNext.Waiting(peer) {
			k.metrics.TotalBeforeExpireWaits.Inc()
			return true
		}
		return false
	}
	depth := k.NeighborhoodDepth()
	for i := range k.commonBinPrefixes {
		binPeersLength := k.knownPeers.BinSize(uint8(i))
		if i >= int(depth) && binPeersLength < len(k.commonBinPrefixes[i]) {
			continue
		}
		binPeers := k.knownPeers.BinPeers(uint8(i))
		for j := range k.commonBinPrefixes[i] {
			pseudoAddr := k.commonBinPrefixes[i][j]
			closestConnectedPeer, err := closestPeer(k.connectedPeers, pseudoAddr, noopSanctionedPeerFn)
			if err != nil {
				if errors.Is(err, topology.ErrNotFound) {
					break
				}
				k.logger.Errorf("closest connected peer: %v", err)
				continue
			}
			closestConnectedPO := swarm.ExtendedProximity(closestConnectedPeer.Bytes(), pseudoAddr.Bytes())
			if int(closestConnectedPO) >= i+k.bitSuffixLength+1 {
				continue
			}
			closestKnownPeer, err := closestPeerInSlice(binPeers, pseudoAddr, skipPeers)
			if err != nil {
				if errors.Is(err, topology.ErrNotFound) {
					break
				}
				k.logger.Errorf("closest known peer: %v", err)
				continue
			}
			if k.connectedPeers.Exists(closestKnownPeer) {
				continue
			}
			closestKnownPeerPO := swarm.ExtendedProximity(closestKnownPeer.Bytes(), pseudoAddr.Bytes())
			if int(closestKnownPeerPO) < i+k.bitSuffixLength+1 {
				continue
			}
			select {
			case <-k.quit:
				return
			default:
				wg.Add(1)
				peerConnChan <- &peerConnInfo{
					po:   swarm.Proximity(k.base.Bytes(), closestKnownPeer.Bytes()),
					addr: closestKnownPeer,
				}
			}
			break
		}
	}
}
func (k *Kad) connectNeighbours(wg *sync.WaitGroup, peerConnChan chan<- *peerConnInfo) {
	sent := 0
	var currentPo uint8 = 0
	_ = k.knownPeers.EachBinRev(func(addr swarm.Address, po uint8) (bool, bool, error) {
		depth := k.NeighborhoodDepth()
		if po < depth {
			return false, true, nil
		}
		if po != currentPo {
			currentPo = po
			sent = 0
		}
		if k.connectedPeers.Exists(addr) {
			return false, false, nil
		}
		if k.waitNext.Waiting(addr) {
			k.metrics.TotalBeforeExpireWaits.Inc()
			return false, false, nil
		}
		select {
		case <-k.quit:
			return true, false, nil
		default:
			wg.Add(1)
			peerConnChan <- &peerConnInfo{
				po:   po,
				addr: addr,
			}
			sent++
		}
		return false, sent == saturationPeers, nil
	})
}
func (k *Kad) connectionAttemptsHandler(ctx context.Context, wg *sync.WaitGroup, neighbourhoodChan, balanceChan <-chan *peerConnInfo) {
	connect := func(peer *peerConnInfo) {
		bzzAddr, err := k.addressBook.Get(peer.addr)
		switch {
		case errors.Is(err, addressbook.ErrNotFound):
			k.logger.Debugf("kademlia: empty address book entry for peer %q", peer.addr)
			k.knownPeers.Remove(peer.addr)
			return
		case err != nil:
			k.logger.Debugf("kademlia: failed to get address book entry for peer %q: %v", peer.addr, err)
			return
		}
		remove := func(peer *peerConnInfo) {
			k.waitNext.Remove(peer.addr)
			k.knownPeers.Remove(peer.addr)
			if err := k.addressBook.Remove(peer.addr); err != nil {
				k.logger.Debugf("kademlia: could not remove peer %q from addressbook", peer.addr)
			}
		}
		switch err = k.connect(ctx, peer.addr, bzzAddr.Underlay); {
		case errors.Is(err, errPruneEntry):
			k.logger.Debugf("kademlia: dial to light node with overlay %q and underlay %q", peer.addr, bzzAddr.Underlay)
			remove(peer)
			return
		case errors.Is(err, errOverlayMismatch):
			k.logger.Debugf("kademlia: overlay mismatch has occurred to an overlay %q with underlay %q", peer.addr, bzzAddr.Underlay)
			remove(peer)
			return
		case err != nil:
			k.logger.Debugf("kademlia: peer not reachable from kademlia %q: %v", bzzAddr, err)
			k.logger.Warningf("peer not reachable when attempting to connect")
			return
		}
		k.waitNext.Set(peer.addr, time.Now().Add(shortRetry), 0)
		k.connectedPeers.Add(peer.addr)
		k.metrics.TotalOutboundConnections.Inc()
		k.collector.Record(peer.addr, im.PeerLogIn(time.Now(), im.PeerConnectionDirectionOutbound))
		k.depthMu.Lock()
		k.depth = recalcDepth(k.connectedPeers, k.radius)
		k.depthMu.Unlock()
		k.logger.Debugf("kademlia: connected to peer: %q in bin: %d", peer.addr, peer.po)
		k.notifyManageLoop()
		k.notifyPeerSig()
	}
	var (
		inProgress   = make(map[string]bool)
		inProgressMu sync.Mutex
	)
	connAttempt := func(peerConnChan <-chan *peerConnInfo) {
		for {
			select {
			case <-k.quit:
				return
			case peer := <-peerConnChan:
				addr := peer.addr.String()
				if k.waitNext.Waiting(peer.addr) {
					k.metrics.TotalBeforeExpireWaits.Inc()
					wg.Done()
					continue
				}
				inProgressMu.Lock()
				if !inProgress[addr] {
					inProgress[addr] = true
					inProgressMu.Unlock()
					connect(peer)
					inProgressMu.Lock()
					delete(inProgress, addr)
				}
				inProgressMu.Unlock()
				wg.Done()
			}
		}
	}
	for i := 0; i < 16; i++ {
		go connAttempt(balanceChan)
	}
	for i := 0; i < 32; i++ {
		go connAttempt(neighbourhoodChan)
	}
}
func (k *Kad) notifyManageLoop() {
	select {
	case k.manageC <- struct{}{}:
	default:
	}
}
func (k *Kad) manage() {
	defer k.wg.Done()
	defer close(k.done)
	defer k.logger.Debugf("kademlia manage loop exited")
	timer := time.NewTimer(0)
	ctx, cancel := context.WithCancel(context.Background())
	go func() {
		<-k.quit
		if !timer.Stop() {
			<-timer.C
		}
		cancel()
	}()
	var wg sync.WaitGroup
	neighbourhoodChan := make(chan *peerConnInfo)
	balanceChan := make(chan *peerConnInfo)
	go k.connectionAttemptsHandler(ctx, &wg, neighbourhoodChan, balanceChan)
	k.wg.Add(1)
	go func() {
		defer k.wg.Done()
		for {
			select {
			case <-k.halt:
				return
			case <-k.quit:
				return
			case <-time.After(5 * time.Minute):
				start := time.Now()
				k.logger.Tracef("kademlia: starting to flush metrics at %s", start)
				if err := k.collector.Flush(); err != nil {
					k.metrics.InternalMetricsFlushTotalErrors.Inc()
					k.logger.Debugf("kademlia: unable to flush metrics counters to the persistent store: %v", err)
				} else {
					k.metrics.InternalMetricsFlushTime.Observe(time.Since(start).Seconds())
					k.logger.Tracef("kademlia: took %s to flush", time.Since(start))
				}
			}
		}
	}()
	k.wg.Add(1)
	go func() {
		defer k.wg.Done()
		for {
			select {
			case <-k.halt:
				return
			case <-k.quit:
				return
			case <-timer.C:
				k.wg.Add(1)
				go func() {
					defer k.wg.Done()
					k.recordPeerLatencies(ctx)
				}()
				_ = timer.Reset(peerPingPollTime)
			}
		}
	}()
	for {
		select {
		case <-k.quit:
			return
		case <-time.After(15 * time.Second):
			k.notifyManageLoop()
		case <-k.manageC:
			start := time.Now()
			select {
			case <-k.halt:
				return
			case <-k.quit:
				return
			default:
			}
			if k.bootnode {
				k.depthMu.Lock()
				depth := k.depth
				radius := k.radius
				k.depthMu.Unlock()
				k.metrics.CurrentDepth.Set(float64(depth))
				k.metrics.CurrentRadius.Set(float64(radius))
				k.metrics.CurrentlyKnownPeers.Set(float64(k.knownPeers.Length()))
				k.metrics.CurrentlyConnectedPeers.Set(float64(k.connectedPeers.Length()))
				continue
			}
			oldDepth := k.NeighborhoodDepth()
			k.connectBalanced(&wg, balanceChan)
			k.connectNeighbours(&wg, neighbourhoodChan)
			wg.Wait()
			k.depthMu.Lock()
			depth := k.depth
			radius := k.radius
			k.depthMu.Unlock()
			k.pruneFunc(depth)
			k.logger.Tracef(
				"kademlia: connector took %s to finish: old depth %d; new depth %d",
				time.Since(start),
				oldDepth,
				depth,
			)
			k.metrics.CurrentDepth.Set(float64(depth))
			k.metrics.CurrentRadius.Set(float64(radius))
			k.metrics.CurrentlyKnownPeers.Set(float64(k.knownPeers.Length()))
			k.metrics.CurrentlyConnectedPeers.Set(float64(k.connectedPeers.Length()))
			if k.connectedPeers.Length() == 0 {
				select {
				case <-k.halt:
					continue
				default:
				}
				k.logger.Debug("kademlia: no connected peers, trying bootnodes")
				k.connectBootNodes(ctx)
			}
		}
	}
}
func (k *Kad) recordPeerLatencies(ctx context.Context) {
	ctx, cancel := context.WithTimeout(ctx, peerPingPollTime)
	defer cancel()
	var wg sync.WaitGroup
	_ = k.connectedPeers.EachBin(func(addr swarm.Address, _ uint8) (bool, bool, error) {
		wg.Add(1)
		go func() {
			defer wg.Done()
			l, err := k.pinger.Ping(ctx, addr, "ping")
			if err != nil {
				k.logger.Tracef("kademlia: cannot get latency for peer %s: %v", addr.String(), err)
				k.blocker.Flag(addr)
				k.metrics.Flag.Inc()
				return
			}
			k.blocker.Unflag(addr)
			k.metrics.Unflag.Inc()
			k.collector.Record(addr, im.PeerLatency(l))
			v := k.collector.Inspect(addr).LatencyEWMA
			k.metrics.PeerLatencyEWMA.Observe(v.Seconds())
		}()
		return false, false, nil
	})
	wg.Wait()
}