func recalcDepth(peers *pslice.PSlice, radius uint8) uint8 {
	if peers.Length() <= nnLowWatermark {
		return 0
	}
	var (
		peersCtr                     = uint(0)
		candidate                    = uint8(0)
		shallowestEmpty, noEmptyBins = peers.ShallowestEmpty()
	)
	shallowestUnsaturated := uint8(0)
	binCount := 0
	_ = peers.EachBinRev(func(_ swarm.Address, bin uint8) (bool, bool, error) {
		if bin == shallowestUnsaturated {
			binCount++
			return false, false, nil
		}
		if bin > shallowestUnsaturated && binCount < quickSaturationPeers {
			return true, false, nil
		}
		shallowestUnsaturated = bin
		binCount = 1
		return false, false, nil
	})
	if !noEmptyBins && shallowestEmpty < shallowestUnsaturated {
		shallowestUnsaturated = shallowestEmpty
	}
	_ = peers.EachBin(func(_ swarm.Address, po uint8) (bool, bool, error) {
		peersCtr++
		if peersCtr >= nnLowWatermark {
			candidate = po
			return true, false, nil
		}
		return false, false, nil
	})
	if shallowestUnsaturated > candidate {
		if radius < candidate {
			return radius
		}
		return candidate
	}
	if radius < shallowestUnsaturated {
		return radius
	}
	return shallowestUnsaturated
}
recalcDepth calculates and returns the kademlia depth.
package kademlia
import (
	"context"
	random "crypto/rand"
	"encoding/json"
	"errors"
	"math"
	"math/big"
	"math/bits"
	"sync"
	"time"
	"github.com/ethsana/sana/pkg/addressbook"
	"github.com/ethsana/sana/pkg/discovery"
	"github.com/ethsana/sana/pkg/logging"
	"github.com/ethsana/sana/pkg/p2p"
	"github.com/ethsana/sana/pkg/shed"
	"github.com/ethsana/sana/pkg/swarm"
	"github.com/ethsana/sana/pkg/topology"
	im "github.com/ethsana/sana/pkg/topology/kademlia/internal/metrics"
	"github.com/ethsana/sana/pkg/topology/kademlia/internal/waitnext"
	"github.com/ethsana/sana/pkg/topology/pslice"
	ma "github.com/multiformats/go-multiaddr"
)
const (
	nnLowWatermark         = 2  
	maxConnAttempts        = 1  
	maxBootNodeAttempts    = 3  
	defaultBitSuffixLength = 3  
	addPeerBatchSize = 500
	peerConnectionAttemptTimeout = 5 * time.Second  
)
var (
	quickSaturationPeers        = 4
	saturationPeers             = 8
	overSaturationPeers         = 20
	bootNodeOverSaturationPeers = 20
	shortRetry                  = 30 * time.Second
	timeToRetry                 = 2 * shortRetry
	broadcastBinSize            = 4
)
var (
	errOverlayMismatch = errors.New("overlay mismatch")
	errPruneEntry      = errors.New("prune entry")
	errEmptyBin        = errors.New("empty bin")
)
type (
	binSaturationFunc  func(bin uint8, peers, connected *pslice.PSlice) (saturated bool, oversaturated bool)
	sanctionedPeerFunc func(peer swarm.Address) bool
)
var noopSanctionedPeerFn = func(_ swarm.Address) bool { return false }
type Options struct {
	SaturationFunc  binSaturationFunc
	Bootnodes       []ma.Multiaddr
	StandaloneMode  bool
	BootnodeMode    bool
	BitSuffixLength int
}
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  
	standalone        bool            
	bootnode          bool            
	collector         *im.Collector
	quit              chan struct{}  
	halt              chan struct{}  
	done              chan struct{}  
	wg                sync.WaitGroup
	waitNext          *waitnext.WaitNext
	metrics           metrics
}
func New(
	base swarm.Address,
	addressbook addressbook.Interface,
	discovery discovery.Driver,
	p2p p2p.Service,
	metricsDB *shed.DB,
	logger logging.Logger,
	o Options,
) *Kad {
	if o.SaturationFunc == nil {
		os := overSaturationPeers
		if o.BootnodeMode {
			os = bootNodeOverSaturationPeers
		}
		o.SaturationFunc = binSaturated(os)
	}
	if o.BitSuffixLength == 0 {
		o.BitSuffixLength = defaultBitSuffixLength
	}
	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,
		standalone:        o.StandaloneMode,
		bootnode:          o.BootnodeMode,
		collector:         im.NewCollector(metricsDB),
		quit:              make(chan struct{}),
		halt:              make(chan struct{}),
		done:              make(chan struct{}),
		wg:                sync.WaitGroup{},
		metrics:           newMetrics(),
	}
	if k.bitSuffixLength > 0 {
		k.generateCommonBinPrefixes()
	}
	return k
}
func (k *Kad) generateCommonBinPrefixes() {
	bitCombinationsCount := int(math.Pow(2, float64(k.bitSuffixLength)))
	bitSufixes := make([]uint8, bitCombinationsCount)
	for i := 0; i < bitCombinationsCount; i++ {
		bitSufixes[i] = uint8(i)
	}
	addr := swarm.MustParseHexAddress(k.base.String())
	addrBytes := addr.Bytes()
	_ = addrBytes
	binPrefixes := k.commonBinPrefixes
	for i := range binPrefixes {
		binPrefixes[i] = make([]swarm.Address, bitCombinationsCount)
	}
	for i := range binPrefixes {
		for j := range binPrefixes[i] {
			pseudoAddrBytes := make([]byte, len(k.base.Bytes()))
			copy(pseudoAddrBytes, k.base.Bytes())
			binPrefixes[i][j] = swarm.NewAddress(pseudoAddrBytes)
		}
	}
	for i := range binPrefixes {
		for j := range binPrefixes[i] {
			pseudoAddrBytes := binPrefixes[i][j].Bytes()
			indexByte, posBit := i/8, i%8
			if hasBit(bits.Reverse8(pseudoAddrBytes[indexByte]), uint8(posBit)) {
				pseudoAddrBytes[indexByte] = bits.Reverse8(clearBit(bits.Reverse8(pseudoAddrBytes[indexByte]), uint8(posBit)))
			} else {
				pseudoAddrBytes[indexByte] = bits.Reverse8(setBit(bits.Reverse8(pseudoAddrBytes[indexByte]), uint8(posBit)))
			}
			bitSuffixPos := k.bitSuffixLength - 1
			for l := i + 1; l < i+k.bitSuffixLength+1; l++ {
				index, pos := l/8, l%8
				if hasBit(bitSufixes[j], uint8(bitSuffixPos)) {
					pseudoAddrBytes[index] = bits.Reverse8(setBit(bits.Reverse8(pseudoAddrBytes[index]), uint8(pos)))
				} else {
					pseudoAddrBytes[index] = bits.Reverse8(clearBit(bits.Reverse8(pseudoAddrBytes[index]), uint8(pos)))
				}
				bitSuffixPos--
			}
			for l := i + k.bitSuffixLength + 1; l < len(pseudoAddrBytes)*8; l++ {
				index, pos := l/8, l%8
				pseudoAddrBytes[index] = bits.Reverse8(clearBit(bits.Reverse8(pseudoAddrBytes[index]), uint8(pos)))
			}
		}
	}
}
func clearBit(n, pos uint8) uint8 {
	mask := ^(uint8(1) << pos)
	return n & mask
}
func setBit(n, pos uint8) uint8 {
	return n | 1<<pos
}
func hasBit(n, pos uint8) bool {
	return n&(1<<pos) > 0
}
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
	}
	for i := range k.commonBinPrefixes {
		if i >= int(k.NeighborhoodDepth()) {
			continue
		}
		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 := closestPeer(k.knownPeers, 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, peerConnChan2 chan<- *peerConnInfo) {
	const multiplePeerThreshold = 8
	sent := 0
	_ = k.knownPeers.EachBinRev(func(addr swarm.Address, po uint8) (bool, bool, error) {
		depth := k.NeighborhoodDepth()
		if depth > po || po >= depth+multiplePeerThreshold {
			return false, true, nil
		}
		if len(k.connectedPeers.BinPeers(po)) >= overSaturationPeers-1 {
			return false, true, nil
		}
		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++
		}
		next := sent == saturationPeers
		if next {
			sent = 0
		}
		return false, next, nil
	})
	_ = k.knownPeers.EachBinRev(func(addr swarm.Address, po uint8) (bool, bool, error) {
		depth := k.NeighborhoodDepth()
		if po < depth+multiplePeerThreshold {
			return false, true, nil
		}
		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)
			peerConnChan2 <- &peerConnInfo{
				po:   po,
				addr: addr,
			}
		}
		return false, true, nil
	})
}
func (k *Kad) connectionAttemptsHandler(ctx context.Context, wg *sync.WaitGroup, peerConnChan, peerConnChan2 <-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 < 64; i++ {
		go connAttempt(peerConnChan)
	}
	for i := 0; i < 8; i++ {
		go connAttempt(peerConnChan2)
	}
}
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")
	ctx, cancel := context.WithCancel(context.Background())
	go func() {
		<-k.quit
		cancel()
	}()
	var wg sync.WaitGroup
	peerConnChan := make(chan *peerConnInfo)
	peerConnChan2 := make(chan *peerConnInfo)
	go k.connectionAttemptsHandler(ctx, &wg, peerConnChan, peerConnChan2)
	for {
		select {
		case <-k.quit:
			return
		case <-time.After(15 * time.Second):
			start := time.Now()
			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(float64(time.Since(start).Nanoseconds()))
			}
			k.notifyManageLoop()
		case <-k.manageC:
			start := time.Now()
			select {
			case <-k.halt:
				return
			case <-k.quit:
				return
			default:
			}
			if k.standalone {
				continue
			}
			oldDepth := k.NeighborhoodDepth()
			k.connectNeighbours(&wg, peerConnChan, peerConnChan2)
			k.connectBalanced(&wg, peerConnChan2)
			wg.Wait()
			k.depthMu.Lock()
			depth := k.depth
			radius := k.radius
			k.depthMu.Unlock()
			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) Start(_ context.Context) error {
	k.wg.Add(1)
	go k.manage()
	go func() {
		select {
		case <-k.halt:
			return
		case <-k.quit:
			return
		default:
		}
		var (
			start     = time.Now()
			addresses []swarm.Address
		)
		err := k.addressBook.IterateOverlays(func(addr swarm.Address) (stop bool, err error) {
			addresses = append(addresses, addr)
			if len(addresses) == addPeerBatchSize {
				k.AddPeers(addresses...)
				addresses = nil
			}
			return false, nil
		})
		if err != nil {
			k.logger.Errorf("addressbook overlays: %w", err)
			return
		}
		k.AddPeers(addresses...)
		k.metrics.StartAddAddressBookOverlaysTime.Observe(float64(time.Since(start).Nanoseconds()))
	}()
	k.notifyManageLoop()
	return nil
}
func (k *Kad) connectBootNodes(ctx context.Context) {
	var attempts, connected int
	totalAttempts := maxBootNodeAttempts * len(k.bootnodes)
	ctx, cancel := context.WithTimeout(ctx, 15*time.Second)
	defer cancel()
	for _, addr := range k.bootnodes {
		if attempts >= totalAttempts || connected >= 3 {
			return
		}
		if _, err := p2p.Discover(ctx, addr, func(addr ma.Multiaddr) (stop bool, err error) {
			k.logger.Tracef("connecting to bootnode %s", addr)
			if attempts >= maxBootNodeAttempts {
				return true, nil
			}
			bzzAddress, err := k.p2p.Connect(ctx, addr)
			attempts++
			k.metrics.TotalBootNodesConnectionAttempts.Inc()
			if err != nil {
				if !errors.Is(err, p2p.ErrAlreadyConnected) {
					k.logger.Debugf("connect fail %s: %v", addr, err)
					k.logger.Warningf("connect to bootnode %s", addr)
					return false, err
				}
				k.logger.Debugf("connect to bootnode fail: %v", err)
				return false, nil
			}
			if err := k.connected(ctx, bzzAddress.Overlay); err != nil {
				return false, err
			}
			k.logger.Tracef("connected to bootnode %s", addr)
			connected++
			return connected >= 3, nil
		}); err != nil && !errors.Is(err, context.Canceled) {
			k.logger.Debugf("discover fail %s: %v", addr, err)
			k.logger.Warningf("discover to bootnode %s", addr)
			return
		}
	}
}
func binSaturated(oversaturationAmount int) binSaturationFunc {
	return func(bin uint8, peers, connected *pslice.PSlice) (bool, bool) {
		potentialDepth := recalcDepth(peers, swarm.MaxPO)
		if bin >= potentialDepth {
			return false, false
		}
		size := 0
		_ = connected.EachBin(func(_ swarm.Address, po uint8) (bool, bool, error) {
			if po == bin {
				size++
			}
			return false, false, nil
		})
		return size >= saturationPeers, size >= oversaturationAmount
	}
}