func (b *blockManager) QueueBlock(block *provautil.Block, sp *serverPeer) {
	if atomic.LoadInt32(&b.shutdown) != 0 {
		sp.blockProcessed <- struct{}{}
		return
	}
	b.msgChan <- &blockMsg{block: block, peer: sp}
}
QueueBlock adds the passed block message and peer to the block handling queue.
package main
import (
	"container/list"
	"net"
	"os"
	"path/filepath"
	"sort"
	"sync"
	"sync/atomic"
	"github.com/bitgo/prova/blockchain"
	"github.com/bitgo/prova/chaincfg"
	"github.com/bitgo/prova/chaincfg/chainhash"
	"github.com/bitgo/prova/database"
	"github.com/bitgo/prova/mempool"
	"github.com/bitgo/prova/provautil"
	"github.com/bitgo/prova/wire"
)
const (
	blockDbNamePrefix = "blocks"
	maxRejectedTxns = 1000
	maxRequestedBlocks = wire.MaxInvPerMsg
	maxRequestedTxns = wire.MaxInvPerMsg
)
var zeroHash chainhash.Hash
type newPeerMsg struct {
	peer *serverPeer
}
type blockMsg struct {
	block *provautil.Block
	peer  *serverPeer
}
type invMsg struct {
	inv  *wire.MsgInv
	peer *serverPeer
}
type donePeerMsg struct {
	peer *serverPeer
}
type txMsg struct {
	tx   *provautil.Tx
	peer *serverPeer
}
type getSyncPeerMsg struct {
	reply chan *serverPeer
}
type processBlockResponse struct {
	isOrphan bool
	err      error
}
type processBlockMsg struct {
	block *provautil.Block
	flags blockchain.BehaviorFlags
	reply chan processBlockResponse
}
type isCurrentMsg struct {
	reply chan bool
}
type pauseMsg struct {
	unpause <-chan struct{}
}
type blockManager struct {
	server          *server
	started         int32
	shutdown        int32
	chain           *blockchain.BlockChain
	rejectedTxns    map[chainhash.Hash]struct{}
	requestedTxns   map[chainhash.Hash]struct{}
	requestedBlocks map[chainhash.Hash]struct{}
	progressLogger  *blockProgressLogger
	syncPeer        *serverPeer
	msgChan         chan interface{}
	wg              sync.WaitGroup
	quit            chan struct{}
}
func (b *blockManager) startSync(peers *list.List) {
	if b.syncPeer != nil {
		return
	}
	best := b.chain.BestSnapshot()
	var bestPeer *serverPeer
	var enext *list.Element
	for e := peers.Front(); e != nil; e = enext {
		enext = e.Next()
		sp := e.Value.(*serverPeer)
		if sp.LastBlock() < best.Height {
			peers.Remove(e)
			continue
		}
		bestPeer = sp
	}
	if bestPeer != nil {
		b.requestedBlocks = make(map[chainhash.Hash]struct{})
		locator, err := b.chain.LatestBlockLocator()
		if err != nil {
			bmgrLog.Errorf("Failed to get block locator for the "+
				"latest block: %v", err)
			return
		}
		bmgrLog.Infof("Syncing to block height %d from peer %v",
			bestPeer.LastBlock(), bestPeer.Addr())
		bestPeer.PushGetBlocksMsg(locator, &zeroHash)
		b.syncPeer = bestPeer
	} else {
		bmgrLog.Warnf("No sync peer candidates available")
	}
}
func (b *blockManager) isSyncCandidate(sp *serverPeer) bool {
	if cfg.RegressionTest {
		host, _, err := net.SplitHostPort(sp.Addr())
		if err != nil {
			return false
		}
		if host != "127.0.0.1" && host != "localhost" {
			return false
		}
	} else {
		if sp.Services()&wire.SFNodeNetwork != wire.SFNodeNetwork {
			return false
		}
	}
	return true
}
func (b *blockManager) handleNewPeerMsg(peers *list.List, sp *serverPeer) {
	if atomic.LoadInt32(&b.shutdown) != 0 {
		return
	}
	bmgrLog.Infof("New valid peer %s (%s)", sp, sp.UserAgent())
	if !b.isSyncCandidate(sp) {
		return
	}
	peers.PushBack(sp)
	b.startSync(peers)
}
func (b *blockManager) handleDonePeerMsg(peers *list.List, sp *serverPeer) {
	for e := peers.Front(); e != nil; e = e.Next() {
		if e.Value == sp {
			peers.Remove(e)
			break
		}
	}
	bmgrLog.Infof("Lost peer %s", sp)
	for k := range sp.requestedTxns {
		delete(b.requestedTxns, k)
	}
	for k := range sp.requestedBlocks {
		delete(b.requestedBlocks, k)
	}
	if b.syncPeer != nil && b.syncPeer == sp {
		b.syncPeer = nil
		b.startSync(peers)
	}
}
func (b *blockManager) handleTxMsg(tmsg *txMsg) {
	txHash := tmsg.tx.Hash()
	if _, exists := b.rejectedTxns[*txHash]; exists {
		bmgrLog.Debugf("Ignoring unsolicited previously rejected "+
			"transaction %v from %s", txHash, tmsg.peer)
		return
	}
	allowOrphans := cfg.MaxOrphanTxs > 0
	acceptedTxs, err := b.server.txMemPool.ProcessTransaction(tmsg.tx,
		allowOrphans, true, mempool.Tag(tmsg.peer.ID()))
	delete(tmsg.peer.requestedTxns, *txHash)
	delete(b.requestedTxns, *txHash)
	if err != nil {
		b.rejectedTxns[*txHash] = struct{}{}
		b.limitMap(b.rejectedTxns, maxRejectedTxns)
		if _, ok := err.(mempool.RuleError); ok {
			bmgrLog.Debugf("Rejected transaction %v from %s: %v",
				txHash, tmsg.peer, err)
		} else {
			bmgrLog.Errorf("Failed to process transaction %v: %v",
				txHash, err)
		}
		code, reason := mempool.ErrToRejectErr(err)
		tmsg.peer.PushRejectMsg(wire.CmdTx, code, reason, txHash,
			false)
		return
	}
	b.server.AnnounceNewTransactions(acceptedTxs)
}
func (b *blockManager) current() bool {
	if !b.chain.IsCurrent() {
		return false
	}
	if b.syncPeer == nil {
		return true
	}
	if b.chain.BestSnapshot().Height < b.syncPeer.LastBlock() {
		return false
	}
	return true
}
func (b *blockManager) handleBlockMsg(bmsg *blockMsg) {
	blockHash := bmsg.block.Hash()
	if _, exists := bmsg.peer.requestedBlocks[*blockHash]; !exists {
		if !cfg.RegressionTest {
			bmgrLog.Warnf("Got unrequested block %v from %s -- "+
				"disconnecting", blockHash, bmsg.peer.Addr())
			bmsg.peer.Disconnect()
			return
		}
	}
	behaviorFlags := blockchain.BFNone
	delete(bmsg.peer.requestedBlocks, *blockHash)
	delete(b.requestedBlocks, *blockHash)
	_, isOrphan, err := b.chain.ProcessBlock(bmsg.block, behaviorFlags)
	if err != nil {
		if _, ok := err.(blockchain.RuleError); ok {
			bmgrLog.Infof("Rejected block %v from %s: %v", blockHash,
				bmsg.peer, err)
		} else {
			bmgrLog.Errorf("Failed to process block %v: %v",
				blockHash, err)
		}
		if dbErr, ok := err.(database.Error); ok && dbErr.ErrorCode ==
			database.ErrCorruption {
			panic(dbErr)
		}
		code, reason := mempool.ErrToRejectErr(err)
		bmsg.peer.PushRejectMsg(wire.CmdBlock, code, reason,
			blockHash, false)
		return
	}
	var heightUpdate uint32
	var blkHashUpdate *chainhash.Hash
	if isOrphan {
		header := &bmsg.block.MsgBlock().Header
		heightUpdate := header.Height
		bmgrLog.Debugf("Extracted height of %v from orphan block", heightUpdate)
		orphanRoot := b.chain.GetOrphanRoot(blockHash)
		locator, err := b.chain.LatestBlockLocator()
		if err != nil {
			bmgrLog.Warnf("Failed to get block locator for the "+
				"latest block: %v", err)
		} else {
			bmsg.peer.PushGetBlocksMsg(locator, orphanRoot)
		}
	} else {
		b.progressLogger.LogBlockHeight(bmsg.block)
		best := b.chain.BestSnapshot()
		heightUpdate = best.Height
		blkHashUpdate = best.Hash
		b.rejectedTxns = make(map[chainhash.Hash]struct{})
		rpcServer := b.server.rpcServer
		if rpcServer != nil {
			rpcServer.gbtWorkState.NotifyBlockConnected(blockHash)
		}
	}
	if blkHashUpdate != nil && heightUpdate != 0 {
		bmsg.peer.UpdateLastBlockHeight(heightUpdate)
		if isOrphan || b.current() {
			go b.server.UpdatePeerHeights(blkHashUpdate, heightUpdate, bmsg.peer)
		}
	}
}
func (b *blockManager) haveInventory(invVect *wire.InvVect) (bool, error) {
	switch invVect.Type {
	case wire.InvTypeBlock:
		return b.chain.HaveBlock(&invVect.Hash)
	case wire.InvTypeTx:
		if b.server.txMemPool.HaveTransaction(&invVect.Hash) {
			return true, nil
		}
		entry, err := b.chain.FetchUtxoEntry(&invVect.Hash)
		if err != nil {
			return false, err
		}
		return entry != nil && !entry.IsFullySpent(), nil
	}
	return true, nil
}
func (b *blockManager) handleInvMsg(imsg *invMsg) {
	lastBlock := -1
	invVects := imsg.inv.InvList
	for i := len(invVects) - 1; i >= 0; i-- {
		if invVects[i].Type == wire.InvTypeBlock {
			lastBlock = i
			break
		}
	}
	if lastBlock != -1 && (imsg.peer != b.syncPeer || b.current()) {
		imsg.peer.UpdateLastAnnouncedBlock(&invVects[lastBlock].Hash)
	}
	if cfg.UseOnlySyncPeerInv && imsg.peer != b.syncPeer && !b.current() {
		return
	}
	if lastBlock != -1 && b.current() {
		blkHeight, err := b.chain.BlockHeightByHash(&invVects[lastBlock].Hash)
		if err == nil {
			imsg.peer.UpdateLastBlockHeight(blkHeight)
		}
	}
	for i, iv := range invVects {
		if iv.Type != wire.InvTypeBlock && iv.Type != wire.InvTypeTx {
			continue
		}
		imsg.peer.AddKnownInventory(iv)
		haveInv, err := b.haveInventory(iv)
		if err != nil {
			bmgrLog.Warnf("Unexpected failure when checking for "+
				"existing inventory during inv message "+
				"processing: %v", err)
			continue
		}
		if !haveInv {
			if iv.Type == wire.InvTypeTx {
				if _, exists := b.rejectedTxns[iv.Hash]; exists {
					continue
				}
			}
			imsg.peer.requestQueue = append(imsg.peer.requestQueue, iv)
			continue
		}
		if iv.Type == wire.InvTypeBlock {
			if b.chain.IsKnownOrphan(&iv.Hash) {
				orphanRoot := b.chain.GetOrphanRoot(&iv.Hash)
				locator, err := b.chain.LatestBlockLocator()
				if err != nil {
					bmgrLog.Errorf("PEER: Failed to get block "+
						"locator for the latest block: "+
						"%v", err)
					continue
				}
				imsg.peer.PushGetBlocksMsg(locator, orphanRoot)
				continue
			}
			if i == lastBlock {
				locator := b.chain.BlockLocatorFromHash(&iv.Hash)
				imsg.peer.PushGetBlocksMsg(locator, &zeroHash)
			}
		}
	}
	numRequested := 0
	gdmsg := wire.NewMsgGetData()
	requestQueue := imsg.peer.requestQueue
	for len(requestQueue) != 0 {
		iv := requestQueue[0]
		requestQueue[0] = nil
		requestQueue = requestQueue[1:]
		switch iv.Type {
		case wire.InvTypeBlock:
			if _, exists := b.requestedBlocks[iv.Hash]; !exists {
				b.requestedBlocks[iv.Hash] = struct{}{}
				b.limitMap(b.requestedBlocks, maxRequestedBlocks)
				imsg.peer.requestedBlocks[iv.Hash] = struct{}{}
				gdmsg.AddInvVect(iv)
				numRequested++
			}
		case wire.InvTypeTx:
			if _, exists := b.requestedTxns[iv.Hash]; !exists {
				b.requestedTxns[iv.Hash] = struct{}{}
				b.limitMap(b.requestedTxns, maxRequestedTxns)
				imsg.peer.requestedTxns[iv.Hash] = struct{}{}
				gdmsg.AddInvVect(iv)
				numRequested++
			}
		}
		if numRequested >= wire.MaxInvPerMsg {
			break
		}
	}
	imsg.peer.requestQueue = requestQueue
	if len(gdmsg.InvList) > 0 {
		imsg.peer.QueueMessage(gdmsg, nil)
	}
}
func (b *blockManager) limitMap(m map[chainhash.Hash]struct{}, limit int) {
	if len(m)+1 > limit {
		for txHash := range m {
			delete(m, txHash)
			return
		}
	}
}
func (b *blockManager) blockHandler() {
	candidatePeers := list.New()
out:
	for {
		select {
		case m := <-b.msgChan:
			switch msg := m.(type) {
			case *newPeerMsg:
				b.handleNewPeerMsg(candidatePeers, msg.peer)
			case *txMsg:
				b.handleTxMsg(msg)
				msg.peer.txProcessed <- struct{}{}
			case *blockMsg:
				b.handleBlockMsg(msg)
				msg.peer.blockProcessed <- struct{}{}
			case *invMsg:
				b.handleInvMsg(msg)
			case *donePeerMsg:
				b.handleDonePeerMsg(candidatePeers, msg.peer)
			case getSyncPeerMsg:
				msg.reply <- b.syncPeer
			case processBlockMsg:
				_, isOrphan, err := b.chain.ProcessBlock(
					msg.block, msg.flags)
				if err != nil {
					msg.reply <- processBlockResponse{
						isOrphan: false,
						err:      err,
					}
				}
				rpcServer := b.server.rpcServer
				if rpcServer != nil {
					rpcServer.gbtWorkState.NotifyBlockConnected(msg.block.Hash())
				}
				msg.reply <- processBlockResponse{
					isOrphan: isOrphan,
					err:      nil,
				}
			case isCurrentMsg:
				msg.reply <- b.current()
			case pauseMsg:
				<-msg.unpause
			default:
				bmgrLog.Warnf("Invalid message type in block "+
					"handler: %T", msg)
			}
		case <-b.quit:
			break out
		}
	}
	b.wg.Done()
	bmgrLog.Trace("Block handler done")
}
func (b *blockManager) handleNotifyMsg(notification *blockchain.Notification) {
	switch notification.Type {
	case blockchain.NTBlockAccepted:
		if !b.current() {
			return
		}
		block, ok := notification.Data.(*provautil.Block)
		if !ok {
			bmgrLog.Warnf("Chain accepted notification is not a block.")
			break
		}
		iv := wire.NewInvVect(wire.InvTypeBlock, block.Hash())
		b.server.RelayInventory(iv, block.MsgBlock().Header)
	case blockchain.NTBlockConnected:
		block, ok := notification.Data.(*provautil.Block)
		if !ok {
			bmgrLog.Warnf("Chain connected notification is not a block.")
			break
		}
		for _, tx := range block.Transactions()[1:] {
			b.server.txMemPool.RemoveTransaction(tx, false)
			b.server.txMemPool.RemoveDoubleSpends(tx)
			b.server.txMemPool.RemoveOrphan(tx)
			acceptedTxs := b.server.txMemPool.ProcessOrphans(tx)
			b.server.AnnounceNewTransactions(acceptedTxs)
		}
		if r := b.server.rpcServer; r != nil {
			for _, tx := range block.Transactions()[1:] {
				iv := wire.NewInvVect(wire.InvTypeTx, tx.Hash())
				b.server.RemoveRebroadcastInventory(iv)
			}
			r.ntfnMgr.NotifyBlockConnected(block)
		}
	case blockchain.NTBlockDisconnected:
		block, ok := notification.Data.(*provautil.Block)
		if !ok {
			bmgrLog.Warnf("Chain disconnected notification is not a block.")
			break
		}
		for _, tx := range block.Transactions()[1:] {
			_, _, err := b.server.txMemPool.MaybeAcceptTransaction(tx,
				false, false)
			if err != nil {
				b.server.txMemPool.RemoveTransaction(tx, true)
			}
		}
		if r := b.server.rpcServer; r != nil {
			r.ntfnMgr.NotifyBlockDisconnected(block)
		}
	}
}
func (b *blockManager) NewPeer(sp *serverPeer) {
	if atomic.LoadInt32(&b.shutdown) != 0 {
		return
	}
	b.msgChan <- &newPeerMsg{peer: sp}
}
func (b *blockManager) QueueTx(tx *provautil.Tx, sp *serverPeer) {
	if atomic.LoadInt32(&b.shutdown) != 0 {
		sp.txProcessed <- struct{}{}
		return
	}
	b.msgChan <- &txMsg{tx: tx, peer: sp}
}