func (dht *FullRT) Provide(ctx context.Context, key cid.Cid, brdcst bool) (err error) {
	ctx, span := Tracer.Start(ctx, "Provide")
	defer span.End()
	if !dht.enableProviders {
		return routing.ErrNotSupported
	} else if !key.Defined() {
		return fmt.Errorf("invalid cid: undefined")
	}
	keyMH := key.Hash()
	logger.Debugw("providing", "cid", key, "mh", internal.LoggableProviderRecordBytes(keyMH))
	dht.ProviderManager.AddProvider(ctx, keyMH, peer.AddrInfo{ID: dht.h.ID()})
	if !brdcst {
		return nil
	}
	closerCtx := ctx
	if deadline, ok := ctx.Deadline(); ok {
		now := time.Now()
		timeout := deadline.Sub(now)
		if timeout < 0 {
			return context.DeadlineExceeded
		} else if timeout < 10*time.Second {
			deadline = deadline.Add(-timeout / 10)
		} else {
			deadline = deadline.Add(-time.Second)
		}
		var cancel context.CancelFunc
		closerCtx, cancel = context.WithDeadline(ctx, deadline)
		defer cancel()
	}
	var exceededDeadline bool
	peers, err := dht.GetClosestPeers(closerCtx, string(keyMH))
	switch err {
	case context.DeadlineExceeded:
		if ctx.Err() != nil {
			return ctx.Err()
		}
		exceededDeadline = true
	case nil:
	default:
		return err
	}
	successes := dht.execOnMany(ctx, func(ctx context.Context, p peer.ID) error {
		err := dht.protoMessenger.PutProvider(ctx, p, keyMH, dht.h)
		return err
	}, peers, true)
	if exceededDeadline {
		return context.DeadlineExceeded
	}
	if successes == 0 {
		return fmt.Errorf("failed to complete provide")
	}
	return ctx.Err()
}
Provider abstraction for indirect stores.
Some DHTs store values directly, while an indirect store stores pointers to
locations of the value, similarly to Coral and Mainline DHT.
Provide makes this node announce that it can provide a value for the given key
package fullrt
import (
	"bytes"
	"context"
	"fmt"
	"math/rand"
	"sync"
	"sync/atomic"
	"time"
	"github.com/multiformats/go-base32"
	"github.com/multiformats/go-multiaddr"
	"github.com/multiformats/go-multihash"
	"github.com/libp2p/go-libp2p-core/host"
	"github.com/libp2p/go-libp2p-core/network"
	"github.com/libp2p/go-libp2p-core/peer"
	"github.com/libp2p/go-libp2p-core/peerstore"
	"github.com/libp2p/go-libp2p-core/protocol"
	"github.com/libp2p/go-libp2p-core/routing"
	"github.com/gogo/protobuf/proto"
	"github.com/ipfs/go-cid"
	ds "github.com/ipfs/go-datastore"
	dssync "github.com/ipfs/go-datastore/sync"
	u "github.com/ipfs/go-ipfs-util"
	logging "github.com/ipfs/go-log"
	kaddht "github.com/libp2p/go-libp2p-kad-dht"
	"github.com/libp2p/go-libp2p-kad-dht/crawler"
	"github.com/libp2p/go-libp2p-kad-dht/internal"
	internalConfig "github.com/libp2p/go-libp2p-kad-dht/internal/config"
	"github.com/libp2p/go-libp2p-kad-dht/internal/net"
	dht_pb "github.com/libp2p/go-libp2p-kad-dht/pb"
	"github.com/libp2p/go-libp2p-kad-dht/providers"
	kb "github.com/libp2p/go-libp2p-kbucket"
	record "github.com/libp2p/go-libp2p-record"
	recpb "github.com/libp2p/go-libp2p-record/pb"
	"github.com/libp2p/go-libp2p-xor/kademlia"
	kadkey "github.com/libp2p/go-libp2p-xor/key"
	"github.com/libp2p/go-libp2p-xor/trie"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/trace"
)
var Tracer = otel.Tracer("")
var logger = logging.Logger("fullrtdht")
type FullRT struct {
	ctx    context.Context
	cancel context.CancelFunc
	wg     sync.WaitGroup
	enableValues, enableProviders bool
	Validator                     record.Validator
	ProviderManager               *providers.ProviderManager
	datastore                     ds.Datastore
	h                             host.Host
	crawlerInterval time.Duration
	lastCrawlTime   time.Time
	crawler        *crawler.Crawler
	protoMessenger *dht_pb.ProtocolMessenger
	messageSender  dht_pb.MessageSender
	filterFromTable kaddht.QueryFilterFunc
	rtLk            sync.RWMutex
	rt              *trie.Trie
	kMapLk       sync.RWMutex
	keyToPeerMap map[string]peer.ID
	peerAddrsLk sync.RWMutex
	peerAddrs   map[peer.ID][]multiaddr.Multiaddr
	bootstrapPeers []*peer.AddrInfo
	bucketSize int
	triggerRefresh chan struct{}
	waitFrac     float64
	timeoutPerOp time.Duration
	bulkSendParallelism int
}
func NewFullRT(h host.Host, protocolPrefix protocol.ID, options ...Option) (*FullRT, error) {
	var fullrtcfg config
	if err := fullrtcfg.apply(options...); err != nil {
		return nil, err
	}
	dhtcfg := &internalConfig.Config{
		Datastore:        dssync.MutexWrap(ds.NewMapDatastore()),
		Validator:        record.NamespacedValidator{},
		ValidatorChanged: false,
		EnableProviders:  true,
		EnableValues:     true,
		ProtocolPrefix:   protocolPrefix,
	}
	if err := dhtcfg.Apply(fullrtcfg.dhtOpts...); err != nil {
		return nil, err
	}
	if err := dhtcfg.ApplyFallbacks(h); err != nil {
		return nil, err
	}
	if err := dhtcfg.Validate(); err != nil {
		return nil, err
	}
	ms := net.NewMessageSenderImpl(h, []protocol.ID{dhtcfg.ProtocolPrefix + "/kad/1.0.0"})
	protoMessenger, err := dht_pb.NewProtocolMessenger(ms)
	if err != nil {
		return nil, err
	}
	c, err := crawler.New(h, crawler.WithParallelism(200))
	if err != nil {
		return nil, err
	}
	ctx, cancel := context.WithCancel(context.Background())
	pm, err := providers.NewProviderManager(ctx, h.ID(), h.Peerstore(), dhtcfg.Datastore)
	if err != nil {
		cancel()
		return nil, err
	}
	var bsPeers []*peer.AddrInfo
	for _, ai := range dhtcfg.BootstrapPeers() {
		tmpai := ai
		bsPeers = append(bsPeers, &tmpai)
	}
	rt := &FullRT{
		ctx:    ctx,
		cancel: cancel,
		enableValues:    dhtcfg.EnableValues,
		enableProviders: dhtcfg.EnableProviders,
		Validator:       dhtcfg.Validator,
		ProviderManager: pm,
		datastore:       dhtcfg.Datastore,
		h:               h,
		crawler:         c,
		messageSender:   ms,
		protoMessenger:  protoMessenger,
		filterFromTable: kaddht.PublicQueryFilter,
		rt:              trie.New(),
		keyToPeerMap:    make(map[string]peer.ID),
		bucketSize:      dhtcfg.BucketSize,
		peerAddrs:      make(map[peer.ID][]multiaddr.Multiaddr),
		bootstrapPeers: bsPeers,
		triggerRefresh: make(chan struct{}),
		waitFrac:     0.3,
		timeoutPerOp: 5 * time.Second,
		crawlerInterval: time.Minute * 60,
		bulkSendParallelism: 20,
	}
	rt.wg.Add(1)
	go rt.runCrawler(ctx)
	return rt, nil
}
type crawlVal struct {
	addrs []multiaddr.Multiaddr
	key   kadkey.Key
}
func (dht *FullRT) TriggerRefresh(ctx context.Context) error {
	select {
	case <-ctx.Done():
		return ctx.Err()
	case dht.triggerRefresh <- struct{}{}:
		return nil
	case <-dht.ctx.Done():
		return fmt.Errorf("dht is closed")
	}
}
func (dht *FullRT) Stat() map[string]peer.ID {
	newMap := make(map[string]peer.ID)
	dht.kMapLk.RLock()
	for k, v := range dht.keyToPeerMap {
		newMap[k] = v
	}
	dht.kMapLk.RUnlock()
	return newMap
}
func (dht *FullRT) Ready() bool {
	dht.rtLk.RLock()
	lastCrawlTime := dht.lastCrawlTime
	dht.rtLk.RUnlock()
	if time.Since(lastCrawlTime) > dht.crawlerInterval {
		return false
	}
	dht.peerAddrsLk.RLock()
	rtSize := len(dht.keyToPeerMap)
	dht.peerAddrsLk.RUnlock()
	return rtSize > len(dht.bootstrapPeers)+1
}
func (dht *FullRT) Host() host.Host {
	return dht.h
}
func (dht *FullRT) runCrawler(ctx context.Context) {
	defer dht.wg.Done()
	t := time.NewTicker(dht.crawlerInterval)
	m := make(map[peer.ID]*crawlVal)
	mxLk := sync.Mutex{}
	initialTrigger := make(chan struct{}, 1)
	initialTrigger <- struct{}{}
	for {
		select {
		case <-t.C:
		case <-initialTrigger:
		case <-dht.triggerRefresh:
		case <-ctx.Done():
			return
		}
		var addrs []*peer.AddrInfo
		dht.peerAddrsLk.Lock()
		for k := range m {
			addrs = append(addrs, &peer.AddrInfo{ID: k})  
		}
		addrs = append(addrs, dht.bootstrapPeers...)
		dht.peerAddrsLk.Unlock()
		for k := range m {
			delete(m, k)
		}
		start := time.Now()
		dht.crawler.Run(ctx, addrs,
			func(p peer.ID, rtPeers []*peer.AddrInfo) {
				conns := dht.h.Network().ConnsToPeer(p)
				var addrs []multiaddr.Multiaddr
				for _, conn := range conns {
					addr := conn.RemoteMultiaddr()
					addrs = append(addrs, addr)
				}
				if len(addrs) == 0 {
					logger.Debugf("no connections to %v after successful query. keeping addresses from the peerstore", p)
					addrs = dht.h.Peerstore().Addrs(p)
				}
				keep := kaddht.PublicRoutingTableFilter(dht, p)
				if !keep {
					return
				}
				mxLk.Lock()
				defer mxLk.Unlock()
				m[p] = &crawlVal{
					addrs: addrs,
				}
			},
			func(p peer.ID, err error) {})
		dur := time.Since(start)
		logger.Infof("crawl took %v", dur)
		peerAddrs := make(map[peer.ID][]multiaddr.Multiaddr)
		kPeerMap := make(map[string]peer.ID)
		newRt := trie.New()
		for k, v := range m {
			v.key = kadkey.KbucketIDToKey(kb.ConvertPeerID(k))
			peerAddrs[k] = v.addrs
			kPeerMap[string(v.key)] = k
			newRt.Add(v.key)
		}
		dht.peerAddrsLk.Lock()
		dht.peerAddrs = peerAddrs
		dht.peerAddrsLk.Unlock()
		dht.kMapLk.Lock()
		dht.keyToPeerMap = kPeerMap
		dht.kMapLk.Unlock()
		dht.rtLk.Lock()
		dht.rt = newRt
		dht.lastCrawlTime = time.Now()
		dht.rtLk.Unlock()
	}
}
func (dht *FullRT) Close() error {
	dht.cancel()
	err := dht.ProviderManager.Process().Close()
	dht.wg.Wait()
	return err
}
func (dht *FullRT) Bootstrap(ctx context.Context) error {
	return nil
}
func (dht *FullRT) CheckPeers(ctx context.Context, peers ...peer.ID) (int, int) {
	var peerAddrs chan interface{}
	var total int
	if len(peers) == 0 {
		dht.peerAddrsLk.RLock()
		total = len(dht.peerAddrs)
		peerAddrs = make(chan interface{}, total)
		for k, v := range dht.peerAddrs {
			peerAddrs <- peer.AddrInfo{
				ID:    k,
				Addrs: v,
			}
		}
		close(peerAddrs)
		dht.peerAddrsLk.RUnlock()
	} else {
		total = len(peers)
		peerAddrs = make(chan interface{}, total)
		dht.peerAddrsLk.RLock()
		for _, p := range peers {
			peerAddrs <- peer.AddrInfo{
				ID:    p,
				Addrs: dht.peerAddrs[p],
			}
		}
		close(peerAddrs)
		dht.peerAddrsLk.RUnlock()
	}
	var success uint64
	workers(100, func(i interface{}) {
		a := i.(peer.AddrInfo)
		dialctx, dialcancel := context.WithTimeout(ctx, time.Second*3)
		if err := dht.h.Connect(dialctx, a); err == nil {
			atomic.AddUint64(&success, 1)
		}
		dialcancel()
	}, peerAddrs)
	return int(success), total
}
func workers(numWorkers int, fn func(interface{}), inputs <-chan interface{}) {
	jobs := make(chan interface{})
	defer close(jobs)
	for i := 0; i < numWorkers; i++ {
		go func() {
			for j := range jobs {
				fn(j)
			}
		}()
	}
	for i := range inputs {
		jobs <- i
	}
}
func (dht *FullRT) GetClosestPeers(ctx context.Context, key string) ([]peer.ID, error) {
	ctx, span := Tracer.Start(ctx, "GetClosestPeers")
	_ = ctx  
	defer span.End()
	kbID := kb.ConvertKey(key)
	kadKey := kadkey.KbucketIDToKey(kbID)
	dht.rtLk.RLock()
	closestKeys := kademlia.ClosestN(kadKey, dht.rt, dht.bucketSize)
	dht.rtLk.RUnlock()
	peers := make([]peer.ID, 0, len(closestKeys))
	for _, k := range closestKeys {
		dht.kMapLk.RLock()
		p, ok := dht.keyToPeerMap[string(k)]
		if !ok {
			logger.Errorf("key not found in map")
		}
		dht.kMapLk.RUnlock()
		dht.peerAddrsLk.RLock()
		peerAddrs := dht.peerAddrs[p]
		dht.peerAddrsLk.RUnlock()
		dht.h.Peerstore().AddAddrs(p, peerAddrs, peerstore.TempAddrTTL)
		peers = append(peers, p)
	}
	return peers, nil
}
func (dht *FullRT) PutValue(ctx context.Context, key string, value []byte, opts ...routing.Option) (err error) {
	if !dht.enableValues {
		return routing.ErrNotSupported
	}
	logger.Debugw("putting value", "key", internal.LoggableRecordKeyString(key))
	if err := dht.Validator.Validate(key, value); err != nil {
		return err
	}
	old, err := dht.getLocal(key)
	if err != nil {
		return err
	}
	if old != nil && !bytes.Equal(old.GetValue(), value) {
		i, err := dht.Validator.Select(key, [][]byte{value, old.GetValue()})
		if err != nil {
			return err
		}
		if i != 0 {
			return fmt.Errorf("can't replace a newer value with an older value")
		}
	}
	rec := record.MakePutRecord(key, value)
	rec.TimeReceived = u.FormatRFC3339(time.Now())
	err = dht.putLocal(key, rec)
	if err != nil {
		return err
	}
	peers, err := dht.GetClosestPeers(ctx, key)
	if err != nil {
		return err
	}
	successes := dht.execOnMany(ctx, func(ctx context.Context, p peer.ID) error {
		routing.PublishQueryEvent(ctx, &routing.QueryEvent{
			Type: routing.Value,
			ID:   p,
		})
		err := dht.protoMessenger.PutValue(ctx, p, rec)
		return err
	}, peers, true)
	if successes == 0 {
		return fmt.Errorf("failed to complete put")
	}
	return nil
}
type RecvdVal struct {
	Val  []byte
	From peer.ID
}
func (dht *FullRT) GetValue(ctx context.Context, key string, opts ...routing.Option) (_ []byte, err error) {
	if !dht.enableValues {
		return nil, routing.ErrNotSupported
	}
	var cfg routing.Options
	if err := cfg.Apply(opts...); err != nil {
		return nil, err
	}
	opts = append(opts, kaddht.Quorum(internalConfig.GetQuorum(&cfg)))
	responses, err := dht.SearchValue(ctx, key, opts...)
	if err != nil {
		return nil, err
	}
	var best []byte
	for r := range responses {
		best = r
	}
	if ctx.Err() != nil {
		return best, ctx.Err()
	}
	if best == nil {
		return nil, routing.ErrNotFound
	}
	logger.Debugf("GetValue %v %x", internal.LoggableRecordKeyString(key), best)
	return best, nil
}
func (dht *FullRT) SearchValue(ctx context.Context, key string, opts ...routing.Option) (<-chan []byte, error) {
	if !dht.enableValues {
		return nil, routing.ErrNotSupported
	}
	var cfg routing.Options
	if err := cfg.Apply(opts...); err != nil {
		return nil, err
	}
	responsesNeeded := 0
	if !cfg.Offline {
		responsesNeeded = internalConfig.GetQuorum(&cfg)
	}
	stopCh := make(chan struct{})
	valCh, lookupRes := dht.getValues(ctx, key, stopCh)
	out := make(chan []byte)
	go func() {
		defer close(out)
		best, peersWithBest, aborted := dht.searchValueQuorum(ctx, key, valCh, stopCh, out, responsesNeeded)
		if best == nil || aborted {
			return
		}
		updatePeers := make([]peer.ID, 0, dht.bucketSize)
		select {
		case l := <-lookupRes:
			if l == nil {
				return
			}
			for _, p := range l.peers {
				if _, ok := peersWithBest[p]; !ok {
					updatePeers = append(updatePeers, p)
				}
			}
		case <-ctx.Done():
			return
		}
		ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
		dht.updatePeerValues(ctx, key, best, updatePeers)
		cancel()
	}()
	return out, nil
}
func (dht *FullRT) searchValueQuorum(ctx context.Context, key string, valCh <-chan RecvdVal, stopCh chan struct{},
	out chan<- []byte, nvals int) ([]byte, map[peer.ID]struct{}, bool) {
	numResponses := 0
	return dht.processValues(ctx, key, valCh,
		func(ctx context.Context, v RecvdVal, better bool) bool {
			numResponses++
			if better {
				select {
				case out <- v.Val:
				case <-ctx.Done():
					return false
				}
			}
			if nvals > 0 && numResponses > nvals {
				close(stopCh)
				return true
			}
			return false
		})
}
func (dht *FullRT) processValues(ctx context.Context, key string, vals <-chan RecvdVal,
	newVal func(ctx context.Context, v RecvdVal, better bool) bool) (best []byte, peersWithBest map[peer.ID]struct{}, aborted bool) {
loop:
	for {
		if aborted {
			return
		}
		select {
		case v, ok := <-vals:
			if !ok {
				break loop
			}
			if best != nil {
				if bytes.Equal(best, v.Val) {
					peersWithBest[v.From] = struct{}{}
					aborted = newVal(ctx, v, false)
					continue
				}
				sel, err := dht.Validator.Select(key, [][]byte{best, v.Val})
				if err != nil {
					logger.Warnw("failed to select best value", "key", internal.LoggableRecordKeyString(key), "error", err)
					continue
				}
				if sel != 1 {
					aborted = newVal(ctx, v, false)
					continue
				}
			}
			peersWithBest = make(map[peer.ID]struct{})
			peersWithBest[v.From] = struct{}{}
			best = v.Val
			aborted = newVal(ctx, v, true)
		case <-ctx.Done():
			return
		}
	}
	return
}
func (dht *FullRT) updatePeerValues(ctx context.Context, key string, val []byte, peers []peer.ID) {
	fixupRec := record.MakePutRecord(key, val)
	for _, p := range peers {
		go func(p peer.ID) {
			if p == dht.h.ID() {
				err := dht.putLocal(key, fixupRec)
				if err != nil {
					logger.Error("Error correcting local dht entry:", err)
				}
				return
			}
			ctx, cancel := context.WithTimeout(ctx, time.Second*5)
			defer cancel()
			err := dht.protoMessenger.PutValue(ctx, p, fixupRec)
			if err != nil {
				logger.Debug("Error correcting DHT entry: ", err)
			}
		}(p)
	}
}
type lookupWithFollowupResult struct {
	peers []peer.ID  
}
func (dht *FullRT) getValues(ctx context.Context, key string, stopQuery chan struct{}) (<-chan RecvdVal, <-chan *lookupWithFollowupResult) {
	valCh := make(chan RecvdVal, 1)
	lookupResCh := make(chan *lookupWithFollowupResult, 1)
	logger.Debugw("finding value", "key", internal.LoggableRecordKeyString(key))
	if rec, err := dht.getLocal(key); rec != nil && err == nil {
		select {
		case valCh <- RecvdVal{
			Val:  rec.GetValue(),
			From: dht.h.ID(),
		}:
		case <-ctx.Done():
		}
	}
	peers, err := dht.GetClosestPeers(ctx, key)
	if err != nil {
		lookupResCh <- &lookupWithFollowupResult{}
		close(valCh)
		close(lookupResCh)
		return valCh, lookupResCh
	}
	go func() {
		defer close(valCh)
		defer close(lookupResCh)
		queryFn := func(ctx context.Context, p peer.ID) error {
			routing.PublishQueryEvent(ctx, &routing.QueryEvent{
				Type: routing.SendingQuery,
				ID:   p,
			})
			rec, peers, err := dht.protoMessenger.GetValue(ctx, p, key)
			if err != nil {
				return err
			}
			routing.PublishQueryEvent(ctx, &routing.QueryEvent{
				Type:      routing.PeerResponse,
				ID:        p,
				Responses: peers,
			})
			if rec == nil {
				return nil
			}
			val := rec.GetValue()
			if val == nil {
				logger.Debug("received a nil record value")
				return nil
			}
			if err := dht.Validator.Validate(key, val); err != nil {
				logger.Debugw("received invalid record (discarded)", "error", err)
				return nil
			}
			select {
			case valCh <- RecvdVal{
				Val:  val,
				From: p,
			}:
			case <-ctx.Done():
				return ctx.Err()
			}
			return nil
		}
		dht.execOnMany(ctx, queryFn, peers, false)
		lookupResCh <- &lookupWithFollowupResult{peers: peers}
	}()
	return valCh, lookupResCh
}