func (lp *LocalPeer) ReceiveInv(ctx context.Context) (*RemotePeer, *wire.MsgInv, error) {
	select {
	case <-ctx.Done():
		return nil, nil, ctx.Err()
	case r := <-lp.receivedInv:
		rp, msg := r.rp, r.msg.(*wire.MsgInv)
		recycleInMsg(r)
		return rp, msg, nil
	}
}
ReceiveInv waits for an inventory message from a remote peer, returning the
peer that sent the message, and the message itself.
package p2p
import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net"
	"net/http"
	"net/url"
	"strconv"
	"sync"
	"sync/atomic"
	"time"
	"decred.org/dcrwallet/v2/errors"
	"decred.org/dcrwallet/v2/lru"
	"decred.org/dcrwallet/v2/version"
	"github.com/decred/dcrd/addrmgr/v2"
	"github.com/decred/dcrd/chaincfg/chainhash"
	"github.com/decred/dcrd/chaincfg/v3"
	"github.com/decred/dcrd/connmgr/v3"
	"github.com/decred/dcrd/gcs/v3"
	blockcf "github.com/decred/dcrd/gcs/v3/blockcf2"
	"github.com/decred/dcrd/wire"
	"golang.org/x/sync/errgroup"
)
const uaName = "dcrwallet"
var uaVersion = version.String()
const minPver = wire.CFilterV2Version
const Pver = wire.InitStateVersion
const maxOutboundConns = 8
const connectTimeout = 30 * time.Second
const stallTimeout = 30 * time.Second
const banThreshold = 100
const invLRUSize = 5000
type msgAck struct {
	msg wire.Message
	ack chan<- struct{}
}
type blockRequest struct {
	hash  *chainhash.Hash
	ready chan struct{}
	block *wire.MsgBlock
	err   error
}
type RemotePeer struct {
	atomicClosed uint64
	id         uint64
	lp         *LocalPeer
	ua         string
	services   wire.ServiceFlag
	pver       uint32
	initHeight int32
	raddr      net.Addr
	na         *addrmgr.NetAddress
	c       net.Conn
	mr      msgReader
	out     chan *msgAck
	outPrio chan *msgAck
	pongs   chan *wire.MsgPong
	requestedBlocksMu sync.Mutex
	requestedBlocks   map[chainhash.Hash]*blockRequest
	requestedCFiltersV2 sync.Map  
	requestedTxs        map[chainhash.Hash]chan<- *wire.MsgTx
	requestedTxsMu      sync.Mutex
	requestedHeaders   chan<- *wire.MsgHeaders  
	sendheaders        bool                     
	requestedHeadersMu sync.Mutex
	invsSent     lru.Cache  
	invsRecv     lru.Cache  
	knownHeaders lru.Cache  
	banScore     connmgr.DynamicBanScore
	err  error          
	errc chan struct{}  
}
type LocalPeer struct {
	atomicMask          uint64
	atomicPeerIDCounter uint64
	atomicRequireHeight int32
	dialer net.Dialer
	receivedGetData  chan *inMsg
	receivedHeaders  chan *inMsg
	receivedInv      chan *inMsg
	announcedHeaders chan *inMsg
	extaddr     net.Addr
	amgr        *addrmgr.AddrManager
	chainParams *chaincfg.Params
	rpByID map[uint64]*RemotePeer
	rpMu   sync.Mutex
}
func NewLocalPeer(params *chaincfg.Params, extaddr *net.TCPAddr, amgr *addrmgr.AddrManager) *LocalPeer {
	lp := &LocalPeer{
		receivedGetData:  make(chan *inMsg),
		receivedHeaders:  make(chan *inMsg),
		receivedInv:      make(chan *inMsg),
		announcedHeaders: make(chan *inMsg),
		extaddr:          extaddr,
		amgr:             amgr,
		chainParams:      params,
		rpByID:           make(map[uint64]*RemotePeer),
	}
	return lp
}
func (lp *LocalPeer) newMsgVersion(pver uint32, extaddr net.Addr, c net.Conn) (*wire.MsgVersion, error) {
	la, err := wire.NewNetAddress(c.LocalAddr(), 0)  
	if err != nil {
		return nil, err
	}
	ra, err := wire.NewNetAddress(c.RemoteAddr(), 0)
	if err != nil {
		return nil, err
	}
	nonce, err := wire.RandomUint64()
	if err != nil {
		return nil, err
	}
	v := wire.NewMsgVersion(la, ra, nonce, 0)
	v.AddUserAgent(uaName, uaVersion)
	v.ProtocolVersion = int32(pver)
	return v, nil
}
func (lp *LocalPeer) RequirePeerHeight(requiredHeight int32) {
	atomic.StoreInt32(&lp.atomicRequireHeight, requiredHeight)
}
func (lp *LocalPeer) ConnectOutbound(ctx context.Context, addr string, reqSvcs wire.ServiceFlag) (*RemotePeer, error) {
	const opf = "localpeer.ConnectOutbound(%v)"
	log.Debugf("Attempting connection to peer %v", addr)
	connectCtx, cancel := context.WithTimeout(ctx, connectTimeout)
	defer cancel()
	id := atomic.AddUint64(&lp.atomicPeerIDCounter, 1)
	tcpAddr, err := net.ResolveTCPAddr("tcp", addr)
	if err != nil {
		return nil, err
	}
	na := addrmgr.NewNetAddressIPPort(tcpAddr.IP, uint16(tcpAddr.Port), wire.SFNodeNetwork)
	na.Timestamp = time.Now()
	rp, err := lp.connectOutbound(connectCtx, id, addr, na)
	if err != nil {
		op := errors.Opf(opf, addr)
		return nil, errors.E(op, err)
	}
	go lp.serveUntilError(ctx, rp)
	var waitForAddrs <-chan time.Time
	if lp.amgr.NeedMoreAddresses() {
		waitForAddrs = time.After(stallTimeout)
		err = rp.Addrs(ctx)
		if err != nil {
			op := errors.Opf(opf, rp.raddr)
			return nil, errors.E(op, err)
		}
	}
	if rp.services&reqSvcs != reqSvcs {
		op := errors.Opf(opf, rp.raddr)
		reason := errors.Errorf("missing required service flags %v", reqSvcs&^rp.services)
		err := errors.E(op, reason)
		go func() {
			if waitForAddrs != nil {
				<-waitForAddrs
			}
			reject := wire.NewMsgReject(wire.CmdVersion, wire.RejectNonstandard, reason.Error())
			rp.sendMessageAck(ctx, reject)
			rp.Disconnect(err)
		}()
		return nil, err
	}
	reqHeight := atomic.LoadInt32(&lp.atomicRequireHeight)
	if rp.initHeight < reqHeight {
		op := errors.Opf(opf, rp.raddr)
		err := errors.E(op, "peer is not synced")
		go func() {
			if waitForAddrs != nil {
				<-waitForAddrs
			}
			rp.Disconnect(err)
		}()
		return nil, err
	}
	lp.amgr.Good(na)
	return rp, nil
}
func (lp *LocalPeer) AddrManager() *addrmgr.AddrManager { return lp.amgr }
func (rp *RemotePeer) NA() *addrmgr.NetAddress { return rp.na }
func (rp *RemotePeer) UA() string { return rp.ua }
func (rp *RemotePeer) ID() uint64 { return rp.id }
func (rp *RemotePeer) InitialHeight() int32 { return rp.initHeight }
func (rp *RemotePeer) Services() wire.ServiceFlag { return rp.services }
func (rp *RemotePeer) InvsSent() *lru.Cache { return &rp.invsSent }
func (rp *RemotePeer) InvsRecv() *lru.Cache { return &rp.invsRecv }
func (rp *RemotePeer) KnownHeaders() *lru.Cache { return &rp.knownHeaders }
func (lp *LocalPeer) SeedPeers(ctx context.Context, services wire.ServiceFlag) {
	seeders := lp.chainParams.Seeders()
	url := &url.URL{
		Scheme:   "https",
		Path:     "/api/addrs",
		RawQuery: fmt.Sprintf("services=%d", services),
	}
	resps := make(chan *http.Response)
	client := http.Client{
		Transport: &http.Transport{
			DialContext: lp.dialer.DialContext,
		},
	}
	cancels := make([]func(), 0, len(seeders))
	defer func() {
		for _, cancel := range cancels {
			cancel()
		}
	}()
	for _, host := range seeders {
		host := host
		url.Host = host
		ctx, cancel := context.WithTimeout(ctx, time.Minute)
		cancels = append(cancels, cancel)
		req, err := http.NewRequestWithContext(ctx, "GET", url.String(), nil)
		if err != nil {
			log.Errorf("Bad seeder request: %v", err)
			continue
		}
		go func() {
			resp, err := client.Do(req)
			if err != nil {
				log.Warnf("Failed to seed addresses from %s: %v", host, err)
				resp = nil
			}
			resps <- resp
		}()
	}
	var na []*addrmgr.NetAddress
	for range seeders {
		resp := <-resps
		if resp == nil {
			continue
		}
		seeder := resp.Request.Host
		var apiResponse struct {
			Host     string `json:"host"`
			Services uint64 `json:"services"`
		}
		dec := json.NewDecoder(io.LimitReader(resp.Body, 4096))
		na = na[:0]
		for i := 0; i < 16; i++ {
			err := dec.Decode(&apiResponse)
			if errors.Is(err, io.EOF) {
				break
			}
			if err != nil {
				log.Warnf("Invalid seeder %v API response: %v", seeder, err)
				break
			}
			host, port, err := net.SplitHostPort(apiResponse.Host)
			if err != nil {
				log.Warnf("Invalid host in seeder %v API: %v", seeder, err)
				continue
			}
			ip := net.ParseIP(host)
			if ip == nil {
				log.Warnf("Invalid IP address %q in seeder %v API host field",
					host, seeder)
				continue
			}
			portNum, err := strconv.ParseUint(port, 10, 16)
			if err != nil {
				log.Warnf("Invalid port %q in seeder %v API host field", port,
					seeder)
				continue
			}
			log.Debugf("Discovered peer %v from seeder %v", apiResponse.Host,
				seeder)
			na = append(na, &addrmgr.NetAddress{
				IP:        ip,
				Port:      uint16(portNum),
				Timestamp: time.Now(),
				Services:  wire.ServiceFlag(apiResponse.Services),
			})
		}
		resp.Body.Close()
		if len(na) > 0 {
			lp.amgr.AddAddresses(na, na[0])
		}
	}
}
type msgReader struct {
	r      io.Reader
	net    wire.CurrencyNet
	msg    wire.Message
	rawMsg []byte
	err    error
}
func (mr *msgReader) next(pver uint32) bool {
	mr.msg, mr.rawMsg, mr.err = wire.ReadMessage(mr.r, pver, mr.net)
	return mr.err == nil
}
func (rp *RemotePeer) writeMessages(ctx context.Context) error {
	e := make(chan error, 1)
	go func() {
		c := rp.c
		pver := rp.pver
		cnet := rp.lp.chainParams.Net
		for {
			var m *msgAck
			select {
			case m = <-rp.outPrio:
			default:
				select {
				case m = <-rp.outPrio:
				case m = <-rp.out:
				}
			}
			log.Debugf("%v -> %v", m.msg.Command(), rp.raddr)
			err := wire.WriteMessage(c, m.msg, pver, cnet)
			if m.ack != nil {
				m.ack <- struct{}{}
			}
			if err != nil {
				e <- err
				return
			}
		}
	}()
	select {
	case <-ctx.Done():
		return ctx.Err()
	case err := <-e:
		return err
	}
}
type msgWriter struct {
	w   io.Writer
	net wire.CurrencyNet
}
func (mw *msgWriter) write(ctx context.Context, msg wire.Message, pver uint32) error {
	e := make(chan error, 1)
	go func() {
		e <- wire.WriteMessage(mw.w, msg, pver, mw.net)
	}()
	select {
	case <-ctx.Done():
		return ctx.Err()
	case err := <-e:
		return err
	}
}
func handshake(ctx context.Context, lp *LocalPeer, id uint64, na *addrmgr.NetAddress, c net.Conn) (*RemotePeer, error) {
	const op errors.Op = "p2p.handshake"
	rp := &RemotePeer{
		id:              id,
		lp:              lp,
		ua:              "",
		services:        0,
		pver:            Pver,
		raddr:           c.RemoteAddr(),
		na:              na,
		c:               c,
		mr:              msgReader{r: c, net: lp.chainParams.Net},
		out:             nil,
		outPrio:         nil,
		pongs:           make(chan *wire.MsgPong, 1),
		requestedBlocks: make(map[chainhash.Hash]*blockRequest),
		requestedTxs:    make(map[chainhash.Hash]chan<- *wire.MsgTx),
		invsSent:        lru.NewCache(invLRUSize),
		invsRecv:        lru.NewCache(invLRUSize),
		knownHeaders:    lru.NewCache(invLRUSize),
		errc:            make(chan struct{}),
	}
	mw := msgWriter{c, lp.chainParams.Net}
	lversion, err := lp.newMsgVersion(rp.pver, lp.extaddr, c)
	if err != nil {
		return nil, errors.E(op, err)
	}
	err = mw.write(ctx, lversion, rp.pver)
	if err != nil {
		return nil, errors.E(op, errors.IO, err)
	}
	err = c.SetReadDeadline(time.Now().Add(3 * time.Second))
	if err != nil {
		return nil, errors.E(op, errors.IO, err)
	}
	msg, _, err := wire.ReadMessage(c, Pver, lp.chainParams.Net)
	if err != nil {
		return nil, errors.E(op, errors.IO, err)
	}
	rversion, ok := msg.(*wire.MsgVersion)
	if !ok {
		return nil, errors.E(op, errors.Protocol, "first received message was not the version message")
	}
	rp.initHeight = rversion.LastBlock
	rp.services = rversion.Services
	rp.ua = rversion.UserAgent
	if uint32(rversion.ProtocolVersion) < minPver {
		return nil, errors.E(op, errors.Protocol, "remote peer has pver lower than minimum required")
	}
	if uint32(rversion.ProtocolVersion) < rp.pver {
		rp.pver = uint32(rversion.ProtocolVersion)
	}
	err = mw.write(ctx, wire.NewMsgVerAck(), rp.pver)
	if err != nil {
		return nil, errors.E(op, errors.IO, err)
	}
	err = c.SetReadDeadline(time.Now().Add(3 * time.Second))
	if err != nil {
		return nil, errors.E(op, errors.IO, err)
	}
	msg, _, err = wire.ReadMessage(c, Pver, lp.chainParams.Net)
	if err != nil {
		return nil, errors.E(op, errors.IO, err)
	}
	_, ok = msg.(*wire.MsgVerAck)
	if !ok {
		return nil, errors.E(op, errors.Protocol, "did not receive verack")
	}
	c.SetReadDeadline(time.Time{})
	rp.out = make(chan *msgAck)
	rp.outPrio = make(chan *msgAck)
	return rp, nil
}
func (lp *LocalPeer) connectOutbound(ctx context.Context, id uint64, addr string,
	na *addrmgr.NetAddress) (*RemotePeer, error) {
	var c net.Conn
	var retryDuration = 5 * time.Second
	timer := time.NewTimer(retryDuration)
	var err error
	for {
		lp.amgr.Attempt(na)
		dialCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
		c, err = lp.dialer.DialContext(dialCtx, "tcp", addr)
		cancel()
		if err == nil {
			break
		}
		var netErr net.Error
		if errors.As(err, &netErr) && !netErr.Temporary() {
			return nil, err
		}
		if ctx.Err() != nil {
			return nil, ctx.Err()
		}
		select {
		case <-ctx.Done():
			timer.Stop()
			return nil, ctx.Err()
		case <-timer.C:
			if retryDuration < 200*time.Second {
				retryDuration += 5 * time.Second
				timer.Reset(retryDuration)
			}
		}
	}
	lp.amgr.Connected(na)
	rp, err := handshake(ctx, lp, id, na, c)
	if err != nil {
		return nil, err
	}
	lp.rpMu.Lock()
	lp.rpByID[rp.id] = rp
	lp.rpMu.Unlock()
	na.Services = rp.services
	return rp, nil
}
func (lp *LocalPeer) serveUntilError(ctx context.Context, rp *RemotePeer) {
	defer func() {
		log.Debugf("Disconnected from outbound peer %v", rp.raddr)
		lp.rpMu.Lock()
		delete(lp.rpByID, rp.id)
		lp.rpMu.Unlock()
	}()
	g, gctx := errgroup.WithContext(ctx)
	g.Go(func() error {
		<-ctx.Done()
		rp.Disconnect(ctx.Err())
		rp.c.Close()
		return nil
	})
	g.Go(func() (err error) {
		defer func() {
			if err != nil && gctx.Err() == nil {
				log.Debugf("remotepeer(%v).readMessages: %v", rp.raddr, err)
			}
		}()
		return rp.readMessages(gctx)
	})
	g.Go(func() (err error) {
		defer func() {
			if err != nil && gctx.Err() == nil {
				log.Debugf("syncWriter(%v).write: %v", rp.raddr, err)
			}
		}()
		return rp.writeMessages(gctx)
	})
	g.Go(func() error {
		for {
			select {
			case <-gctx.Done():
				return gctx.Err()
			case <-time.After(2 * time.Minute):
				ctx, cancel := context.WithDeadline(gctx, time.Now().Add(15*time.Second))
				rp.pingPong(ctx)
				cancel()
			}
		}
	})
	err := g.Wait()
	if err != nil {
		rp.Disconnect(err)
	}
}
var ErrDisconnected = errors.New("peer has been disconnected")
func (rp *RemotePeer) Disconnect(reason error) {
	if !atomic.CompareAndSwapUint64(&rp.atomicClosed, 0, 1) {
		return
	}
	log.Debugf("Disconnecting %v", rp.raddr)
	rp.c.Close()
	if reason == nil {
		reason = ErrDisconnected
	}
	rp.err = reason
	close(rp.errc)
}
func (rp *RemotePeer) Err() error {
	<-rp.errc
	return rp.err
}
func (rp *RemotePeer) RemoteAddr() net.Addr {
	return rp.c.RemoteAddr()
}
func (rp *RemotePeer) LocalAddr() net.Addr {
	return rp.c.LocalAddr()
}
func (rp *RemotePeer) BanScore() uint32 {
	return rp.banScore.Int()
}
func (rp *RemotePeer) Pver() uint32 { return rp.pver }
func (rp *RemotePeer) String() string {
	return rp.raddr.String()
}
type inMsg struct {
	rp  *RemotePeer
	msg wire.Message
}
var inMsgPool = sync.Pool{
	New: func() interface{} { return new(inMsg) },
}
func newInMsg(rp *RemotePeer, msg wire.Message) *inMsg {
	m := inMsgPool.Get().(*inMsg)
	m.rp = rp
	m.msg = msg
	return m
}
func recycleInMsg(m *inMsg) {
	*m = inMsg{}
	inMsgPool.Put(m)
}
func (rp *RemotePeer) readMessages(ctx context.Context) error {
	for rp.mr.next(rp.pver) {
		msg := rp.mr.msg
		log.Debugf("%v <- %v", msg.Command(), rp.raddr)
		if _, ok := msg.(*wire.MsgVersion); ok {
			return errors.E(errors.Protocol, "received unexpected version message")
		}
		go func() {
			switch m := msg.(type) {
			case *wire.MsgAddr:
				rp.receivedAddr(ctx, m)
			case *wire.MsgBlock:
				rp.receivedBlock(ctx, m)
			case *wire.MsgCFilterV2:
				rp.receivedCFilterV2(ctx, m)
			case *wire.MsgNotFound:
				rp.receivedNotFound(ctx, m)
			case *wire.MsgTx:
				rp.receivedTx(ctx, m)
			case *wire.MsgGetData:
				rp.receivedGetData(ctx, m)
			case *wire.MsgHeaders:
				rp.receivedHeaders(ctx, m)
			case *wire.MsgInv:
				if rp.lp.messageIsMasked(MaskInv) {
					rp.lp.receivedInv <- newInMsg(rp, msg)
				}
			case *wire.MsgReject:
				log.Warnf("%v reject(%v, %v, %v): %v", rp.raddr, m.Cmd, m.Code, &m.Hash, m.Reason)
			case *wire.MsgGetMiningState:
				rp.receivedGetMiningState(ctx)
			case *wire.MsgGetInitState:
				rp.receivedGetInitState(ctx)
			case *wire.MsgPing:
				pong(ctx, m, rp)
			case *wire.MsgPong:
				rp.receivedPong(ctx, m)
			}
		}()
	}
	return rp.mr.err
}
func pong(ctx context.Context, ping *wire.MsgPing, rp *RemotePeer) {
	ctx, cancel := context.WithTimeout(ctx, 5*time.Minute)
	defer cancel()
	select {
	case <-ctx.Done():
	case rp.outPrio <- &msgAck{wire.NewMsgPong(ping.Nonce), nil}:
	}
}
type MessageMask uint64
const (
	MaskGetData MessageMask = 1 << iota
	MaskInv
)
func (lp *LocalPeer) AddHandledMessages(mask MessageMask) {
	for {
		p := atomic.LoadUint64(&lp.atomicMask)
		n := p | uint64(mask)
		if atomic.CompareAndSwapUint64(&lp.atomicMask, p, n) {
			return
		}
	}
}
func (lp *LocalPeer) RemoveHandledMessages(mask MessageMask) {
	for {
		p := atomic.LoadUint64(&lp.atomicMask)
		n := p &^ uint64(mask)
		if atomic.CompareAndSwapUint64(&lp.atomicMask, p, n) {
			return
		}
	}
}
func (lp *LocalPeer) messageIsMasked(m MessageMask) bool {
	return atomic.LoadUint64(&lp.atomicMask)&uint64(m) != 0
}
func (lp *LocalPeer) ReceiveGetData(ctx context.Context) (*RemotePeer, *wire.MsgGetData, error) {
	select {
	case <-ctx.Done():
		return nil, nil, ctx.Err()
	case r := <-lp.receivedGetData:
		rp, msg := r.rp, r.msg.(*wire.MsgGetData)
		recycleInMsg(r)
		return rp, msg, nil
	}
}