func (s *Syncer) receiveGetData(ctx context.Context) error {
	var wg sync.WaitGroup
	for {
		rp, msg, err := s.lp.ReceiveGetData(ctx)
		if err != nil {
			wg.Wait()
			return err
		}
		wg.Add(1)
		go func() {
			defer wg.Done()
			var txHashes []*chainhash.Hash
			var notFound []*wire.InvVect
			for _, inv := range msg.InvList {
				if !rp.InvsSent().Contains(inv.Hash) {
					notFound = append(notFound, inv)
					continue
				}
				switch inv.Type {
				case wire.InvTypeTx:
					txHashes = append(txHashes, &inv.Hash)
				default:
					notFound = append(notFound, inv)
				}
			}
			var foundTxs []*wire.MsgTx
			if len(txHashes) != 0 {
				var missing []*wire.InvVect
				var err error
				foundTxs, missing, err = s.wallet.GetTransactionsByHashes(ctx, txHashes)
				if err != nil && !errors.Is(err, errors.NotExist) {
					log.Warnf("Failed to look up transactions for getdata reply to peer %v: %v",
						rp.RemoteAddr(), err)
					return
				}
				for _, miss := range missing {
					if v, ok := s.mempool.Load(miss.Hash); ok {
						tx := v.(*wire.MsgTx)
						foundTxs = append(foundTxs, tx)
						continue
					}
					notFound = append(notFound, miss)
				}
			}
			for _, tx := range foundTxs {
				err := rp.SendMessage(ctx, tx)
				if ctx.Err() != nil {
					return
				}
				if err != nil {
					log.Warnf("Failed to send getdata reply to peer %v: %v",
						rp.RemoteAddr(), err)
				}
			}
			if len(notFound) != 0 {
				err := rp.SendMessage(ctx, &wire.MsgNotFound{InvList: notFound})
				if ctx.Err() != nil {
					return
				}
				if err != nil {
					log.Warnf("Failed to send notfound reply to peer %v: %v",
						rp.RemoteAddr(), err)
				}
			}
		}()
	}
}
receiveGetData handles all received getdata requests from peers.  An inv
message declaring knowledge of the data must have been previously sent to the
peer, or a notfound message reports the data as missing.  Only transactions
may be queried by a peer.
package spv
import (
	"context"
	"runtime"
	"sync"
	"sync/atomic"
	"time"
	"decred.org/dcrwallet/v2/errors"
	"decred.org/dcrwallet/v2/lru"
	"decred.org/dcrwallet/v2/p2p"
	"decred.org/dcrwallet/v2/validate"
	"decred.org/dcrwallet/v2/wallet"
	"github.com/decred/dcrd/addrmgr/v2"
	"github.com/decred/dcrd/blockchain/stake/v4"
	"github.com/decred/dcrd/chaincfg/chainhash"
	"github.com/decred/dcrd/gcs/v3/blockcf2"
	"github.com/decred/dcrd/wire"
	"golang.org/x/sync/errgroup"
)
const reqSvcs = wire.SFNodeNetwork
type Syncer struct {
	atomicCatchUpTryLock uint32  
	atomicWalletSynced   uint32  
	wallet *wallet.Wallet
	lp     *p2p.LocalPeer
	discoverAccounts bool
	loadedFilters    bool
	persistentPeers []string
	connectingRemotes map[string]struct{}
	remotes           map[string]*p2p.RemotePeer
	remotesMu         sync.Mutex
	rescanFilter *wallet.RescanFilter
	filterData   blockcf2.Entries
	filterMu     sync.Mutex
	seenTxs lru.Cache
	sidechains  wallet.SidechainForest
	sidechainMu sync.Mutex
	currentLocators   []*chainhash.Hash
	locatorGeneration uint
	locatorMu         sync.Mutex
	notifications *Notifications
	mempool     sync.Map  
	mempoolAdds chan *chainhash.Hash
}
type Notifications struct {
	Synced                       func(sync bool)
	PeerConnected                func(peerCount int32, addr string)
	PeerDisconnected             func(peerCount int32, addr string)
	FetchMissingCFiltersStarted  func()
	FetchMissingCFiltersProgress func(startCFiltersHeight, endCFiltersHeight int32)
	FetchMissingCFiltersFinished func()
	FetchHeadersStarted          func()
	FetchHeadersProgress         func(lastHeaderHeight int32, lastHeaderTime int64)
	FetchHeadersFinished         func()
	DiscoverAddressesStarted     func()
	DiscoverAddressesFinished    func()
	RescanStarted                func()
	RescanProgress               func(rescannedThrough int32)
	RescanFinished               func()
	MempoolTxs func(txs []*wire.MsgTx)
	TipChanged func(tip *wire.BlockHeader, reorgDepth int32, txs []*wire.MsgTx)
}
func NewSyncer(w *wallet.Wallet, lp *p2p.LocalPeer) *Syncer {
	return &Syncer{
		wallet:            w,
		discoverAccounts:  !w.Locked(),
		connectingRemotes: make(map[string]struct{}),
		remotes:           make(map[string]*p2p.RemotePeer),
		rescanFilter:      wallet.NewRescanFilter(nil, nil),
		seenTxs:           lru.NewCache(2000),
		lp:                lp,
		mempoolAdds:       make(chan *chainhash.Hash),
	}
}
func (s *Syncer) SetPersistentPeers(peers []string) {
	s.persistentPeers = peers
}
func (s *Syncer) SetNotifications(ntfns *Notifications) {
	s.notifications = ntfns
}
func (s *Syncer) DisableDiscoverAccounts() {
	s.discoverAccounts = false
}
func (s *Syncer) synced() {
	if atomic.CompareAndSwapUint32(&s.atomicWalletSynced, 0, 1) &&
		s.notifications != nil &&
		s.notifications.Synced != nil {
		s.notifications.Synced(true)
	}
}
func (s *Syncer) Synced() bool {
	return atomic.LoadUint32(&s.atomicWalletSynced) == 1
}
func (s *Syncer) EstimateMainChainTip() int32 {
	_, chainTip := s.wallet.MainChainTip(context.Background())
	s.forRemotes(func(rp *p2p.RemotePeer) error {
		if rp.InitialHeight() > chainTip {
			chainTip = rp.InitialHeight()
		}
		return nil
	})
	return chainTip
}
func (s *Syncer) GetRemotePeers() map[string]*p2p.RemotePeer {
	s.remotesMu.Lock()
	defer s.remotesMu.Unlock()
	remotes := make(map[string]*p2p.RemotePeer, len(s.remotes))
	for k, rp := range s.remotes {
		remotes[k] = rp
	}
	return remotes
}
func (s *Syncer) unsynced() {
	if atomic.CompareAndSwapUint32(&s.atomicWalletSynced, 1, 0) &&
		s.notifications != nil &&
		s.notifications.Synced != nil {
		s.notifications.Synced(false)
	}
}
func (s *Syncer) peerConnected(remotesCount int, addr string) {
	if s.notifications != nil && s.notifications.PeerConnected != nil {
		s.notifications.PeerConnected(int32(remotesCount), addr)
	}
}
func (s *Syncer) peerDisconnected(remotesCount int, addr string) {
	if s.notifications != nil && s.notifications.PeerDisconnected != nil {
		s.notifications.PeerDisconnected(int32(remotesCount), addr)
	}
}
func (s *Syncer) fetchMissingCfiltersStart() {
	if s.notifications != nil && s.notifications.FetchMissingCFiltersStarted != nil {
		s.notifications.FetchMissingCFiltersStarted()
	}
}
func (s *Syncer) fetchMissingCfiltersProgress(startMissingCFilterHeight, endMissinCFilterHeight int32) {
	if s.notifications != nil && s.notifications.FetchMissingCFiltersProgress != nil {
		s.notifications.FetchMissingCFiltersProgress(startMissingCFilterHeight, endMissinCFilterHeight)
	}
}
func (s *Syncer) fetchMissingCfiltersFinished() {
	if s.notifications != nil && s.notifications.FetchMissingCFiltersFinished != nil {
		s.notifications.FetchMissingCFiltersFinished()
	}
}
func (s *Syncer) fetchHeadersStart() {
	if s.notifications != nil && s.notifications.FetchHeadersStarted != nil {
		s.notifications.FetchHeadersStarted()
	}
}
func (s *Syncer) fetchHeadersProgress(lastHeader *wire.BlockHeader) {
	if s.notifications != nil && s.notifications.FetchHeadersProgress != nil {
		s.notifications.FetchHeadersProgress(int32(lastHeader.Height), lastHeader.Timestamp.Unix())
	}
}
func (s *Syncer) fetchHeadersFinished() {
	if s.notifications != nil && s.notifications.FetchHeadersFinished != nil {
		s.notifications.FetchHeadersFinished()
	}
}
func (s *Syncer) discoverAddressesStart() {
	if s.notifications != nil && s.notifications.DiscoverAddressesStarted != nil {
		s.notifications.DiscoverAddressesStarted()
	}
}
func (s *Syncer) discoverAddressesFinished() {
	if s.notifications != nil && s.notifications.DiscoverAddressesFinished != nil {
		s.notifications.DiscoverAddressesFinished()
	}
}
func (s *Syncer) rescanStart() {
	if s.notifications != nil && s.notifications.RescanStarted != nil {
		s.notifications.RescanStarted()
	}
}
func (s *Syncer) rescanProgress(rescannedThrough int32) {
	if s.notifications != nil && s.notifications.RescanProgress != nil {
		s.notifications.RescanProgress(rescannedThrough)
	}
}
func (s *Syncer) rescanFinished() {
	if s.notifications != nil && s.notifications.RescanFinished != nil {
		s.notifications.RescanFinished()
	}
}
func (s *Syncer) mempoolTxs(txs []*wire.MsgTx) {
	if s.notifications != nil && s.notifications.MempoolTxs != nil {
		s.notifications.MempoolTxs(txs)
	}
}
func (s *Syncer) tipChanged(tip *wire.BlockHeader, reorgDepth int32, matchingTxs map[chainhash.Hash][]*wire.MsgTx) {
	if s.notifications != nil && s.notifications.TipChanged != nil {
		var txs []*wire.MsgTx
		for _, matching := range matchingTxs {
			txs = append(txs, matching...)
		}
		s.notifications.TipChanged(tip, reorgDepth, txs)
	}
}
func (s *Syncer) setRequiredHeight(tipHeight int32) {
	requireHeight := tipHeight
	if requireHeight > 6 {
		requireHeight -= 6
	}
	s.lp.RequirePeerHeight(requireHeight)
}
func (s *Syncer) Run(ctx context.Context) error {
	tipHash, tipHeight := s.wallet.MainChainTip(ctx)
	s.setRequiredHeight(tipHeight)
	rescanPoint, err := s.wallet.RescanPoint(ctx)
	if err != nil {
		return err
	}
	log.Infof("Headers synced through block %v height %d", &tipHash, tipHeight)
	if rescanPoint != nil {
		h, err := s.wallet.BlockHeader(ctx, rescanPoint)
		if err != nil {
			return err
		}
		log.Infof("Transactions synced through block %v height %d", &h.PrevBlock, h.Height-1)
	} else {
		log.Infof("Transactions synced through block %v height %d", &tipHash, tipHeight)
	}
	locators, err := s.wallet.BlockLocators(ctx, nil)
	if err != nil {
		return err
	}
	s.currentLocators = locators
	s.lp.AddrManager().Start()
	defer func() {
		err := s.lp.AddrManager().Stop()
		if err != nil {
			log.Errorf("Failed to cleanly stop address manager: %v", err)
		}
	}()
	if len(s.persistentPeers) == 0 {
		s.lp.SeedPeers(ctx, wire.SFNodeNetwork|wire.SFNodeCF)
	}
	g, ctx := errgroup.WithContext(ctx)
	g.Go(func() error { return s.receiveGetData(ctx) })
	g.Go(func() error { return s.receiveInv(ctx) })
	g.Go(func() error { return s.receiveHeadersAnnouncements(ctx) })
	s.lp.AddHandledMessages(p2p.MaskGetData | p2p.MaskInv)
	if len(s.persistentPeers) != 0 {
		for i := range s.persistentPeers {
			raddr := s.persistentPeers[i]
			g.Go(func() error { return s.connectToPersistent(ctx, raddr) })
		}
	} else {
		g.Go(func() error { return s.connectToCandidates(ctx) })
	}
	g.Go(func() error { return s.handleMempool(ctx) })
	s.wallet.SetNetworkBackend(s)
	defer s.wallet.SetNetworkBackend(nil)
	return g.Wait()
}
func (s *Syncer) peerCandidate(svcs wire.ServiceFlag) (*addrmgr.NetAddress, error) {
	for tries := 0; tries < 100; tries++ {
		kaddr := s.lp.AddrManager().GetAddress()
		if kaddr == nil {
			break
		}
		na := kaddr.NetAddress()
		k := na.Key()
		s.remotesMu.Lock()
		_, isConnecting := s.connectingRemotes[k]
		_, isRemote := s.remotes[k]
		switch {
		case isConnecting || isRemote:
			fallthrough
		case tries < 30 && time.Since(kaddr.LastAttempt()) < 10*time.Minute:
			fallthrough
		case tries < 50 && kaddr.NetAddress().Services&svcs != svcs:
			s.remotesMu.Unlock()
			continue
		}
		s.connectingRemotes[k] = struct{}{}
		s.remotesMu.Unlock()
		return na, nil
	}
	return nil, errors.New("no addresses")
}
func (s *Syncer) connectToPersistent(ctx context.Context, raddr string) error {
	for {
		func() {
			ctx, cancel := context.WithCancel(ctx)
			defer cancel()
			rp, err := s.lp.ConnectOutbound(ctx, raddr, reqSvcs)
			if err != nil {
				if ctx.Err() == nil {
					log.Errorf("Peering attempt failed: %v", err)
				}
				return
			}
			log.Infof("New peer %v %v %v", raddr, rp.UA(), rp.Services())
			k := rp.NA().Key()
			s.remotesMu.Lock()
			s.remotes[k] = rp
			n := len(s.remotes)
			s.remotesMu.Unlock()
			s.peerConnected(n, k)
			wait := make(chan struct{})
			go func() {
				err := s.startupSync(ctx, rp)
				if err != nil {
					rp.Disconnect(err)
				}
				wait <- struct{}{}
			}()
			err = rp.Err()
			s.remotesMu.Lock()
			delete(s.remotes, k)
			n = len(s.remotes)
			s.remotesMu.Unlock()
			s.peerDisconnected(n, k)
			<-wait
			if ctx.Err() != nil {
				return
			}
			log.Warnf("Lost peer %v: %v", raddr, err)
		}()
		if err := ctx.Err(); err != nil {
			return err
		}
		time.Sleep(5 * time.Second)
	}
}
func (s *Syncer) connectToCandidates(ctx context.Context) error {
	var wg sync.WaitGroup
	defer wg.Wait()
	sem := make(chan struct{}, 8)
	for {
		if ctx.Err() != nil {
			return ctx.Err()
		}
		select {
		case sem <- struct{}{}:
		case <-ctx.Done():
			return ctx.Err()
		}
		na, err := s.peerCandidate(reqSvcs)
		if err != nil {
			select {
			case <-ctx.Done():
				return ctx.Err()
			case <-time.After(5 * time.Second):
				<-sem
				continue
			}
		}
		wg.Add(1)
		go func() {
			ctx, cancel := context.WithCancel(ctx)
			defer func() {
				cancel()
				wg.Done()
				<-sem
			}()
			raddr := na.String()
			rp, err := s.lp.ConnectOutbound(ctx, raddr, reqSvcs)
			if err != nil {
				s.remotesMu.Lock()
				delete(s.connectingRemotes, raddr)
				s.remotesMu.Unlock()
				if ctx.Err() == nil {
					log.Warnf("Peering attempt failed: %v", err)
				}
				return
			}
			log.Infof("New peer %v %v %v", raddr, rp.UA(), rp.Services())
			s.remotesMu.Lock()
			delete(s.connectingRemotes, raddr)
			s.remotes[raddr] = rp
			n := len(s.remotes)
			s.remotesMu.Unlock()
			s.peerConnected(n, raddr)
			wait := make(chan struct{})
			go func() {
				err := s.startupSync(ctx, rp)
				if err != nil {
					rp.Disconnect(err)
				}
				wait <- struct{}{}
			}()
			err = rp.Err()
			if ctx.Err() != context.Canceled {
				log.Warnf("Lost peer %v: %v", raddr, err)
			}
			<-wait
			s.remotesMu.Lock()
			delete(s.remotes, raddr)
			n = len(s.remotes)
			s.remotesMu.Unlock()
			s.peerDisconnected(n, raddr)
		}()
	}
}
func (s *Syncer) forRemotes(f func(rp *p2p.RemotePeer) error) error {
	defer s.remotesMu.Unlock()
	s.remotesMu.Lock()
	if len(s.remotes) == 0 {
		return errors.E(errors.NoPeers)
	}
	for _, rp := range s.remotes {
		err := f(rp)
		if err != nil {
			return err
		}
	}
	return nil
}
func (s *Syncer) pickRemote(pick func(*p2p.RemotePeer) bool) (*p2p.RemotePeer, error) {
	defer s.remotesMu.Unlock()
	s.remotesMu.Lock()
	for _, rp := range s.remotes {
		if pick(rp) {
			return rp, nil
		}
	}
	return nil, errors.E(errors.NoPeers)
}