service.go

/*
 * Copyright (C) 2018 The ontology Authors
 * This file is part of The ontology library.
 *
 * The ontology is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * The ontology is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public License
 * along with The ontology.  If not, see <http://www.gnu.org/licenses/>.
 */

package vbft

import (
	"bytes"
	"fmt"
	"math"
	"reflect"
	"sync"
	"time"

	"github.com/ontio/ontology-crypto/keypair"
	"github.com/ontio/ontology-crypto/vrf"
	"github.com/ontio/ontology-eventbus/actor"
	"github.com/qbyyf/ontology/account"
	"github.com/qbyyf/ontology/common"
	"github.com/qbyyf/ontology/common/log"
	actorTypes "github.com/qbyyf/ontology/consensus/actor"
	vconfig "github.com/qbyyf/ontology/consensus/vbft/config"
	"github.com/qbyyf/ontology/core/ledger"
	"github.com/qbyyf/ontology/core/payload"
	"github.com/qbyyf/ontology/core/types"
	"github.com/qbyyf/ontology/core/utils"
	"github.com/qbyyf/ontology/events"
	"github.com/qbyyf/ontology/events/message"
	p2pmsg "github.com/qbyyf/ontology/p2pserver/message/types"
	p2p "github.com/qbyyf/ontology/p2pserver/net/protocol"
	gover "github.com/qbyyf/ontology/smartcontract/service/native/governance"
	ninit "github.com/qbyyf/ontology/smartcontract/service/native/init"
	nutils "github.com/qbyyf/ontology/smartcontract/service/native/utils"
	"github.com/qbyyf/ontology/validator/increment"
)

type BftActionType uint8

const (
	MakeProposal BftActionType = iota
	EndorseBlock
	CommitBlock
	SealBlock
	FastForward // for syncer catch up
	ReBroadcast
	SubmitBlock
)

const (
	CAP_MESSAGE_CHANNEL  = 4096
	CAP_ACTION_CHANNEL   = 64
	CAP_MSG_SEND_CHANNEL = 16
)

type BftAction struct {
	Type     BftActionType
	BlockNum uint32
	Proposal *blockProposalMsg
	forEmpty bool
}

type BlockParticipantConfig struct {
	BlockNum    uint32
	Vrf         vconfig.VRFValue
	ChainConfig *vconfig.ChainConfig
	Proposers   []uint32
	Endorsers   []uint32
	Committers  []uint32
}

type p2pMsgPayload struct {
	fromPeer uint32
	payload  *p2pmsg.ConsensusPayload
}

type Server struct {
	Index         uint32
	account       *account.Account
	poolActor     *actorTypes.TxPoolActor
	p2p           p2p.P2P
	ledger        *ledger.Ledger
	incrValidator *increment.IncrementValidator
	pid           *actor.PID

	// some config
	msgHistoryDuration uint32

	//
	// Note:
	// 1. locking priority: metaLock > blockpool.Lock > peerpool.Lock
	// 2. should never take exclusive lock on both blockpool and peerpool at the same time.
	// 3. msgpool.Lock is independent, should have no exclusive overlap with other locks.
	//
	metaLock                 sync.RWMutex
	completedBlockNum        uint32 // ledger SaveBlockCompleted block num
	currentBlockNum          uint32
	LastConfigBlockNum       uint32
	config                   *vconfig.ChainConfig
	currentParticipantConfig *BlockParticipantConfig

	chainStore *ChainStore // block store
	msgPool    *MsgPool    // consensus msg pool
	blockPool  *BlockPool  // received block proposals
	peerPool   *PeerPool   // consensus peers
	syncer     *Syncer
	stateMgr   *StateMgr
	timer      *EventTimer

	msgRecvC   *sync.Map // map[uint32]chan *p2pMsgPayload
	msgC       chan ConsensusMsg
	bftActionC chan *BftAction
	msgSendC   chan *SendMsgEvent
	sub        *events.ActorSubscriber
	quitC      chan struct{}
	quit       bool
	quitWg     sync.WaitGroup
}

func NewVbftServer(account *account.Account, txpool *actor.PID, p2p p2p.P2P) (*Server, error) {
	server := &Server{
		msgHistoryDuration: 64,
		account:            account,
		poolActor:          &actorTypes.TxPoolActor{Pool: txpool},
		p2p:                p2p,
		ledger:             ledger.DefLedger,
		incrValidator:      increment.NewIncrementValidator(20),
	}
	server.stateMgr = newStateMgr(server)

	props := actor.FromProducer(func() actor.Actor {
		return server
	})

	pid, err := actor.SpawnNamed(props, "consensus_vbft")
	if err != nil {
		return nil, err
	}
	server.pid = pid
	server.sub = events.NewActorSubscriber(pid)

	if err := server.initialize(); err != nil {
		return nil, fmt.Errorf("vbft server start failed: %s", err)
	}
	return server, nil
}

func (self *Server) Receive(context actor.Context) {
	switch msg := context.Message().(type) {
	case *actor.Restarting:
		log.Info("vbft actor restarting")
	case *actor.Stopping:
		log.Info("vbft actor stopping")
	case *actor.Stopped:
		log.Info("vbft actor stopped")
	case *actor.Started:
		log.Info("vbft actor started")
	case *actor.Restart:
		log.Info("vbft actor restart")
	case *actorTypes.StartConsensus:
		log.Info("vbft actor start consensus")
	case *actorTypes.StopConsensus:
		self.stop()
	case *message.SaveBlockCompleteMsg:
		log.Infof("vbft actor SaveBlockCompleteMsg receives block complete event. block height=%d, numtx=%d",
			msg.Block.Header.Height, len(msg.Block.Transactions))
		self.handleBlockPersistCompleted(msg.Block)
	case *message.BlockConsensusComplete:
		log.Infof("vbft actor  BlockConsensusComplete receives block complete event. block height=%d, numtx=%d",
			msg.Block.Header.Height, len(msg.Block.Transactions))
		self.handleBlockPersistCompleted(msg.Block)
	case *p2pmsg.ConsensusPayload:
		self.NewConsensusPayload(msg)

	default:
		log.Info("vbft actor: Unknown msg ", msg, "type", reflect.TypeOf(msg))
	}
}

func (self *Server) GetPID() *actor.PID {
	return self.pid
}

func (self *Server) Start() error {
	return self.start()
}

func (self *Server) Halt() error {
	self.pid.Tell(&actorTypes.StopConsensus{})
	return nil
}

func (self *Server) handleBlockPersistCompleted(block *types.Block) {
	log.Infof("persist block: %d, %x", block.Header.Height, block.Hash())

	if block.Header.Height <= self.GetCompletedBlockNum() {
		log.Infof("server %d, persist block %d, vs completed %d",
			self.Index, block.Header.Height, self.GetCompletedBlockNum())
		return
	}
	self.SetCompletedBlockNum(block.Header.Height)
	self.incrValidator.AddBlock(block)
	if self.nonConsensusNode() {
		self.blockPool.ReloadFromLedger()
		if self.GetCommittedBlockNo() >= self.GetCurrentBlockNo() {
			self.SetCurrentBlockNo(self.GetCommittedBlockNo() + 1)
		}
	}

	if self.checkNeedUpdateChainConfig(self.GetCompletedBlockNum()) || self.checkUpdateChainConfig(self.GetCompletedBlockNum()) {
		err := self.updateChainConfig()
		if err != nil {
			log.Errorf("updateChainConfig failed:%s", err)
		}
	}
}

func (self *Server) CheckSubmitBlock(blkNum uint32, stateRoot common.Uint256) bool {
	cMsgs := self.msgPool.GetBlockSubmitMsgNums(blkNum)
	var stateRootCnt uint32
	for _, msg := range cMsgs {
		c := msg.(*blockSubmitMsg)
		if c != nil {
			if c.BlockStateRoot == stateRoot {
				stateRootCnt++
			} else {
				continue
			}
		}
	}

	cfg := self.GetChainConfig()
	m := cfg.N - (cfg.N-1)/3

	if stateRootCnt < uint32(m) {
		return false
	}
	return true
}

func (self *Server) NewConsensusPayload(payload *p2pmsg.ConsensusPayload) {
	peerID := vconfig.PubkeyID(payload.Owner)
	peerIdx, present := self.peerPool.GetPeerIndex(peerID)
	if !present {
		log.Debugf("invalid consensus node: %s", peerID)
		return
	}
	if self.peerPool.isNewPeer(peerIdx) {
		self.peerPool.peerConnected(peerIdx)
	}
	p2pid, present := self.peerPool.getP2pId(peerIdx)
	if !present || p2pid != payload.PeerId {
		self.peerPool.addP2pId(peerIdx, payload.PeerId)
	}

	if C := self.GetPeerMsgChan(peerIdx); C != nil {
		C <- &p2pMsgPayload{
			fromPeer: peerIdx,
			payload:  payload,
		}
	} else {
		log.Errorf("consensus msg without receiver: %d node: %s", peerIdx, peerID)
		return
	}
}

func (self *Server) LoadChainConfig(blkNum uint32) error {
	//get chainconfig from genesis block

	block, _ := self.blockPool.getSealedBlock(blkNum)
	if block == nil {
		return fmt.Errorf("getSealedBlock err height:%d", blkNum)
	}
	var cfg vconfig.ChainConfig
	if block.getNewChainConfig() != nil {
		cfg = *block.getNewChainConfig()
		self.LastConfigBlockNum = block.getLastConfigBlockNum()
	} else {
		cfgBlock := block
		if block.getLastConfigBlockNum() != math.MaxUint32 {
			cfgBlock, _ = self.blockPool.getSealedBlock(block.getLastConfigBlockNum())
			if cfgBlock == nil {
				return fmt.Errorf("failed to get cfg block height:%d", block.getLastConfigBlockNum())
			}
		}
		if cfgBlock.getNewChainConfig() == nil {
			panic("failed to get chain config from config block")
		}
		cfg = *cfgBlock.getNewChainConfig()
		self.LastConfigBlockNum = cfgBlock.getLastConfigBlockNum()
	}
	self.metaLock.Lock()
	self.config = &cfg
	self.metaLock.Unlock()

	self.metaLock.RLock()
	defer self.metaLock.RUnlock()

	if self.config.View == 0 || self.config.MaxBlockChangeView == 0 {
		panic("invalid view or maxblockchangeview ")
	}

	// update timer params
	self.updateTimerParams(self.config)

	// protected by server.metaLock
	self.SetCompletedBlockNum(self.GetCommittedBlockNo())
	self.SetCurrentBlockNo(self.GetCommittedBlockNo() + 1)

	log.Infof("committed: %d, current block no: %d", self.GetCommittedBlockNo(), self.GetCurrentBlockNo())

	block, _ = self.blockPool.getSealedBlock(self.GetCommittedBlockNo())
	if block == nil {
		return fmt.Errorf("failed to get sealed block (%d)", self.GetCommittedBlockNo())
	}
	currentParticipantConfig, err := self.buildParticipantConfig(self.GetCurrentBlockNo(), block, self.config)
	if err != nil {
		return fmt.Errorf("failed to build participant config: %s", err)
	}
	self.currentParticipantConfig = currentParticipantConfig

	return nil
}

func (self *Server) nonConsensusNode() bool {
	return self.Index == math.MaxUint32
}

//updateChainCofig
func (self *Server) updateChainConfig() error {
	block, _ := self.blockPool.getSealedBlock(self.GetCompletedBlockNum())
	if block == nil {
		return fmt.Errorf("GetBlockInfo failed,block is nil:%d", self.GetCompletedBlockNum())
	}
	if block.Info.NewChainConfig == nil {
		return fmt.Errorf("GetNewChainConfig nil,%d", self.GetCompletedBlockNum())
	}
	log.Infof("updateChainConfig blkNum:%d", self.GetCompletedBlockNum())
	self.metaLock.Lock()
	self.config = block.Info.NewChainConfig
	self.LastConfigBlockNum = block.getLastConfigBlockNum()
	self.metaLock.Unlock()

	self.metaLock.RLock()
	defer self.metaLock.RUnlock()

	// . update timer param
	// . update peer pool
	// . remove nonparticipation consensus node
	// . update statemgr peers
	// . reset remove peer connections, create new connections with new peers
	self.updateTimerParams(self.config)

	pubkey := vconfig.PubkeyID(self.account.PublicKey)
	peermap := make(map[uint32]string)
	for _, p := range self.GetChainConfig().Peers {
		peermap[p.Index] = p.ID
		if self.Index == math.MaxUint32 && pubkey == p.ID {
			self.Index = p.Index
			log.Infof("updateChainConfig add index :%d", self.Index)
		}
		_, present := self.peerPool.GetPeerIndex(p.ID)
		if !present {
			// check if peer pubkey support VRF
			if pk, err := vconfig.Pubkey(p.ID); err != nil {
				return fmt.Errorf("failed to parse peer %d PeerID: %s", p.Index, err)
			} else if !vrf.ValidatePublicKey(pk) {
				return fmt.Errorf("peer %d: invalid peer pubkey for VRF", p.Index)
			}

			if err := self.peerPool.addPeer(p); err != nil {
				return fmt.Errorf("failed to add peer %d: %s", p.Index, err)
			}
			publickey, err := vconfig.Pubkey(p.ID)
			if err != nil {
				log.Errorf("Pubkey failed: %v", err)
				return fmt.Errorf("Pubkey failed: %v", err)
			}
			peerIdx := p.Index
			self.CreatePeerMsgChan(peerIdx)
			go func() {
				if err := self.run(publickey); err != nil {
					log.Errorf("server %d, processor on peer %d failed: %s",
						self.Index, peerIdx, err)
				}
			}()
			log.Infof("updateChainConfig add peer index:%v,id:%v", p.ID, p.Index)
		}
	}
	for index, peerPubKey := range self.peerPool.GetAllPubKeys() {
		_, present := peermap[index]
		if !present {
			if index == self.Index {
				self.Index = math.MaxUint32
				log.Infof("updateChainConfig remove index :%d", index)
			} else {
				if C := self.GetPeerMsgChan(index); C != nil {
					pubkey := vconfig.PubkeyID(peerPubKey)
					self.peerPool.RemovePeerIndex(pubkey)
					log.Infof("updateChainConfig remove consensus:index:%d,id:%v", index, pubkey)
					C <- nil
				}
			}
		}
	}
	return nil
}

func (self *Server) initialize() error {
	// TODO: load config from chain

	// TODO: configurable log
	selfNodeId := vconfig.PubkeyID(self.account.PublicKey)
	log.Infof("server: %s starting", selfNodeId)

	store, err := OpenBlockStore(self.ledger, self.pid)
	if err != nil {
		log.Errorf("failed to open block store: %s", err)
		return fmt.Errorf("failed to open block store: %s", err)
	}
	self.chainStore = store
	log.Info("block store opened")

	self.blockPool, err = newBlockPool(self, self.msgHistoryDuration, store)
	if err != nil {
		log.Errorf("init blockpool: %s", err)
		return fmt.Errorf("init blockpool: %s", err)
	}
	self.msgPool = newMsgPool(self, self.msgHistoryDuration)
	self.peerPool = NewPeerPool(0, self) // FIXME: maxSize
	self.timer = NewEventTimer(self)
	self.syncer = newSyncer(self)

	self.msgRecvC = new(sync.Map)
	self.msgC = make(chan ConsensusMsg, CAP_MESSAGE_CHANNEL)
	self.bftActionC = make(chan *BftAction, CAP_ACTION_CHANNEL)
	self.msgSendC = make(chan *SendMsgEvent, CAP_MSG_SEND_CHANNEL)

	self.quitC = make(chan struct{})
	if err := self.LoadChainConfig(store.GetChainedBlockNum()); err != nil {
		log.Errorf("failed to load config: %s", err)
		return fmt.Errorf("failed to load config: %s", err)
	}
	log.Infof("chain config loaded from local, current blockNum: %d", self.GetCurrentBlockNo())

	// add all consensus peers to peer_pool
	for _, p := range self.GetChainConfig().Peers {
		// check if peer pubkey support VRF
		if pk, err := vconfig.Pubkey(p.ID); err != nil {
			return fmt.Errorf("failed to parse peer %d PeerID: %s", p.Index, err)
		} else if !vrf.ValidatePublicKey(pk) {
			return fmt.Errorf("peer %d: invalid peer pubkey for VRF", p.Index)
		}

		if err := self.peerPool.addPeer(p); err != nil {
			return fmt.Errorf("failed to add peer %d: %s", p.Index, err)
		}
		log.Infof("added peer: %s", p.ID)
	}

	//index equal math.MaxUint32  is noconsensus node
	id := vconfig.PubkeyID(self.account.PublicKey)
	index, present := self.peerPool.GetPeerIndex(id)
	if present {
		self.Index = index
	} else {
		self.Index = math.MaxUint32
	}
	self.sub.Subscribe(message.TOPIC_SAVE_BLOCK_COMPLETE)
	go self.syncer.run()
	go self.stateMgr.run()
	go self.msgSendLoop()
	go self.timerLoop()
	go self.actionLoop()
	self.quitWg.Add(1)
	go func() {
		defer self.quitWg.Done()

		for {
			if err := self.processMsgEvent(); err != nil {
				log.Errorf("server %d: %s", self.Index, err)
			}
			if self.quit {
				break
			}
		}
	}()

	self.stateMgr.StateEventC <- &StateEvent{
		Type: ConfigLoaded,
	}

	log.Infof("peer %d started", self.Index)

	// TODO: start peer-conn-handlers

	return nil
}

func (self *Server) start() error {
	// check if server pubkey support VRF
	if !vrf.ValidatePrivateKey(self.account.PrivateKey) || !vrf.ValidatePublicKey(self.account.PublicKey) {
		return fmt.Errorf("server %d consensus start failed: invalid account key for VRF", self.Index)
	}

	// start heartbeat ticker
	self.timer.startPeerTicker(math.MaxUint32)

	// start peers msg handlers
	for _, p := range self.GetChainConfig().Peers {
		peerIdx := p.Index
		pk := self.peerPool.GetPeerPubKey(peerIdx)
		self.CreatePeerMsgChan(peerIdx)

		go func() {
			if err := self.run(pk); err != nil {
				log.Errorf("server %d, processor on peer %d failed: %s",
					self.Index, peerIdx, err)
			}
		}()
	}

	return nil
}

func (self *Server) stop() {

	self.incrValidator.Clean()
	self.sub.Unsubscribe(message.TOPIC_SAVE_BLOCK_COMPLETE)
	// stop syncer, statemgr, msgSendLoop, timer, actionLoop, msgProcessingLoop
	self.quit = true
	close(self.quitC)
	self.quitWg.Wait()

	self.syncer.stop()
	self.timer.stop()
	self.msgPool.clean()
	self.blockPool.clean()
	self.chainStore.close()
	self.peerPool.clean()
}

//
// go routine per net connection
//
func (self *Server) run(peerPubKey keypair.PublicKey) error {
	peerID := vconfig.PubkeyID(peerPubKey)
	peerIdx, present := self.peerPool.GetPeerIndex(peerID)
	if !present {
		return fmt.Errorf("invalid consensus node: %s", peerID)
	}

	// broadcast heartbeat
	self.heartbeat()

	// wait remote msgs
	self.peerPool.waitPeerConnected(peerIdx)

	defer func() {
		// TODO: handle peer disconnection here
		log.Warnf("server %d: disconnected with peer %d", self.Index, peerIdx)
		self.ClosePeerMsgChan(peerIdx)

		self.peerPool.peerDisconnected(peerIdx)
		self.stateMgr.StateEventC <- &StateEvent{
			Type: UpdatePeerState,
			peerState: &PeerState{
				peerIdx:   peerIdx,
				connected: false,
			},
		}
	}()

	errC := make(chan error)
	go func() {
		for {
			fromPeer, msgData, err := self.receiveFromPeer(peerIdx)
			if err != nil {
				errC <- err
				return
			}
			msg, err := DeserializeVbftMsg(msgData)

			if err != nil {
				log.Errorf("server %d failed to deserialize vbft msg (len %d): %s", self.Index, len(msgData), err)
			} else {
				pk := self.peerPool.GetPeerPubKey(fromPeer)
				if pk == nil {
					log.Errorf("server %d failed to get peer %d pubkey", self.Index, fromPeer)
					continue
				}

				if msg.Type() == BlockProposalMessage {
					if proposal := msg.(*blockProposalMsg); proposal != nil {
						fromPeer = proposal.Block.getProposer()
						pk = self.peerPool.GetPeerPubKey(proposal.Block.getProposer())
					}
				}

				if err := msg.Verify(pk); err != nil {
					log.Errorf("server %d failed to verify msg, type %d, err: %s",
						self.Index, msg.Type(), err)
					continue
				}

				if msg.Type() <= BlockCommitMessage || msg.Type() == BlockSubmitMessage {
					log.Infof("server %d received consensus msg, blk %d, type: %d from %d",
						self.Index, msg.GetBlockNum(), msg.Type(), fromPeer)
				}

				self.onConsensusMsg(fromPeer, msg, hashData(msgData))
			}
		}
	}()

	return <-errC
}

func (self *Server) getState() ServerState {
	return self.stateMgr.getState()
}

func (self *Server) updateParticipantConfig() error {
	blkNum := self.GetCurrentBlockNo()
	block, _ := self.blockPool.getSealedBlock(blkNum - 1)
	if block == nil {
		return fmt.Errorf("failed to get sealed block (%d)", blkNum-1)
	}

	chainconfig := block.Info.NewChainConfig
	if chainconfig == nil {
		chainCfg := self.GetChainConfig()
		chainconfig = &chainCfg
	}
	cfg, err := self.buildParticipantConfig(blkNum, block, chainconfig)
	if err != nil {
		return fmt.Errorf("failed to build participant config (%d): %s", blkNum, err)
	}

	self.metaLock.Lock()
	self.currentParticipantConfig = cfg
	self.metaLock.Unlock()

	return nil
}

func (self *Server) startNewRound() error {
	blkNum := self.GetCurrentBlockNo()

	if err := self.updateParticipantConfig(); err != nil {
		log.Errorf("startNewRound error:%s", err)
		return err
	}
	// check proposals in msgpool
	var proposal *blockProposalMsg
	if proposals := self.msgPool.GetProposalMsgs(blkNum); len(proposals) > 0 {
		for _, p := range proposals {
			msg := p.(*blockProposalMsg)
			if msg == nil {
				continue
			}
			if self.isProposer(blkNum, msg.Block.getProposer()) {
				// get proposal from proposer, process it
				proposal = msg
			} else {
				// add other proposals to blockpool
				if err := self.blockPool.newBlockProposal(msg); err != nil {
					log.Errorf("starting new round, failed to add proposal from %d: %s",
						msg.Block.getProposer(), err)
				}
			}
		}
	}

	endorses := self.msgPool.GetEndorsementsMsgs(blkNum)
	if len(endorses) > 0 {
		for _, e := range endorses {
			msg := e.(*blockEndorseMsg)
			if msg == nil {
				continue
			}
			self.blockPool.newBlockEndorsement(msg)
		}
	}

	commits := self.msgPool.GetCommitMsgs(blkNum)
	if len(commits) > 0 {
		for _, c := range commits {
			msg := c.(*blockCommitMsg)
			if msg == nil {
				continue
			}
			if err := self.blockPool.newBlockCommitment(msg); err != nil {
				log.Infof("start new round, failed to add commit, blk %d, commit for %d: %s",
					blkNum, msg.BlockProposer, err)
			}
		}
	}

	chainCfg := self.GetChainConfig()
	if _, _, done := self.blockPool.commitDone(blkNum, chainCfg.C, chainCfg.N); done && len(commits) > 0 {
		// resend commit msg to msg-processor to restart commit-done processing
		// Note: commitDone will set Done flag in block-pool, so removed Done flag checking
		// in commit msg processing.
		self.blockPool.setCommitDone(blkNum)
		self.processConsensusMsg(commits[0])
		return nil
	} else if _, _, done := self.blockPool.endorseDone(blkNum, chainCfg.C); done && len(endorses) > 0 {
		// resend endorse msg to msg-processor to restart endorse-done processing
		self.processConsensusMsg(endorses[0])
		return nil
	} else if proposal != nil {
		self.processProposalMsg(proposal)
		return nil
	}
	if err := self.timer.startTxTicker(blkNum); err != nil {
		log.Errorf("startxticker blk:%d,err:%s", blkNum, err)
		return err
	}
	if err := self.timer.StartTxBlockTimeout(blkNum); err != nil {
		log.Errorf("starttxblocktimeout blk:%d,err:%s", blkNum, err)
		return err
	}
	return nil
}

func (self *Server) startNewProposal(blkNum uint32) {
	// make proposal
	if self.isProposer(blkNum, self.Index) {
		log.Infof("server %d, proposer for block %d", self.Index, blkNum)
		// FIXME: possible deadlock on channel
		self.bftActionC <- &BftAction{
			Type:     MakeProposal,
			BlockNum: blkNum,
			forEmpty: false,
		}
	} else if self.is2ndProposer(blkNum, self.Index) {
		log.Infof("server %d, 2nd proposer for block %d", self.Index, blkNum)
		if err := self.timer.StartProposalBackoffTimer(blkNum); err != nil {
			log.Errorf("server %d, startproposalbackofftimer for block %d err:%s", self.Index, blkNum, err)
		}
	}

	// TODO: if new round block proposal has received, go endorsing/committing directly

	if err := self.timer.StartProposalTimer(blkNum); err != nil {
		log.Errorf("server %d, startnewproposal for block %d err:%s", self.Index, blkNum, err)
	}
}

// verify consensus messsage, then send msg to processMsgEvent
func (self *Server) onConsensusMsg(peerIdx uint32, msg ConsensusMsg, msgHash common.Uint256) {

	if self.msgPool.HasMsg(msg, msgHash) && msg.Type() != BlockCommitMessage {
		// dup msg checking
		log.Debugf("dup msg with msg type %d from %d", msg.Type(), peerIdx)
		return
	}

	switch msg.Type() {
	case BlockProposalMessage:
		pMsg, ok := msg.(*blockProposalMsg)
		if !ok {
			log.Error("invalid msg with proposal msg type")
			return
		}

		msgBlkNum := pMsg.GetBlockNum()
		if msgBlkNum > self.GetCurrentBlockNo() {
			// for concurrency, support two active consensus round
			if err := self.msgPool.AddMsg(msg, msgHash); err != nil {
				if err != errDropFarFutureMsg {
					log.Errorf("failed to add proposal msg (%d) to pool: %s", msgBlkNum, err)
				}
				return
			}

			if isReady(self.getState()) {
				// set the peer as syncing-check trigger from current round
				// start syncing check from current round
				self.syncer.syncCheckReqC <- &SyncCheckReq{
					msg:      msg,
					blockNum: self.GetCurrentBlockNo(),
				}
			}
		} else if msgBlkNum < self.GetCurrentBlockNo() {

			if msgBlkNum <= self.GetCommittedBlockNo() {
				if msgBlkNum+MAX_SYNCING_CHECK_BLK_NUM < self.GetCommittedBlockNo() {
					log.Infof("server %d get proposal msg for block %d, from %d, current committed %d",
						self.Index, msgBlkNum, pMsg.Block.getProposer(), self.GetCommittedBlockNo())
					self.timer.C <- &TimerEvent{
						evtType:  EventPeerHeartbeat,
						blockNum: pMsg.Block.getProposer(),
					}
				}
				return
			}

			if isReady(self.getState()) {
				// start syncing check from proposed block round
				self.syncer.syncCheckReqC <- &SyncCheckReq{
					msg:      msg,
					blockNum: msgBlkNum,
				}
			}

		} else {
			if err := self.msgPool.AddMsg(msg, msgHash); err != nil {
				log.Errorf("failed to add proposal msg (%d) to pool", msgBlkNum)
				return
			}
			self.processProposalMsg(pMsg)
		}

	case BlockEndorseMessage:
		pMsg, ok := msg.(*blockEndorseMsg)
		if !ok {
			log.Error("invalid msg with endorse msg type")
			return
		}

		// TODO: verify msg

		msgBlkNum := pMsg.GetBlockNum()
		if msgBlkNum > self.GetCurrentBlockNo() {
			// for concurrency, support two active consensus round
			if err := self.msgPool.AddMsg(msg, msgHash); err != nil {
				if err != errDropFarFutureMsg {
					log.Errorf("failed to add endorse msg (%d) to pool: %s", msgBlkNum, err)
				}
				return
			}

			if isReady(self.getState()) {
				// set the peer as syncing-check trigger from current round
				// start syncing check from current round
				self.syncer.syncCheckReqC <- &SyncCheckReq{
					msg:      msg,
					blockNum: self.GetCurrentBlockNo(),
				}
			}
		} else if msgBlkNum < self.GetCurrentBlockNo() {
			if msgBlkNum <= self.GetCommittedBlockNo() {
				if msgBlkNum+MAX_SYNCING_CHECK_BLK_NUM < self.GetCommittedBlockNo() {
					log.Infof("server %d get endorse msg for block %d, from %d, current committed %d",
						self.Index, msgBlkNum, pMsg.Endorser, self.GetCommittedBlockNo())
					self.timer.C <- &TimerEvent{
						evtType:  EventPeerHeartbeat,
						blockNum: pMsg.Endorser,
					}
				}
				return
			}

			if isReady(self.getState()) {
				// start syncing check from proposed block round
				self.syncer.syncCheckReqC <- &SyncCheckReq{
					msg:      msg,
					blockNum: msgBlkNum,
				}
			}

		} else {
			// add to msg pool
			if err := self.msgPool.AddMsg(msg, msgHash); err != nil {
				log.Errorf("failed to add endorse msg (%d) to pool", msgBlkNum)
				return
			}
			self.processConsensusMsg(msg)
		}

	case BlockCommitMessage:
		pMsg, ok := msg.(*blockCommitMsg)
		if !ok {
			log.Error("invalid msg with commit msg type")
			return
		}

		// TODO: verify msg

		msgBlkNum := pMsg.GetBlockNum()
		if msgBlkNum > self.GetCurrentBlockNo() {
			if err := self.msgPool.AddMsg(msg, msgHash); err != nil {
				if err != errDropFarFutureMsg {
					log.Errorf("failed to add commit msg (%d) to pool: %s", msgBlkNum, err)
				}
				return
			}

			if isReady(self.getState()) {
				// set the peer as syncing-check trigger from current round
				// start syncing check from current round
				self.syncer.syncCheckReqC <- &SyncCheckReq{
					msg:      msg,
					blockNum: self.GetCurrentBlockNo(),
				}
			}

		} else if msgBlkNum < self.GetCurrentBlockNo() {
			if msgBlkNum <= self.GetCommittedBlockNo() {
				if msgBlkNum+MAX_SYNCING_CHECK_BLK_NUM < self.GetCommittedBlockNo() {
					log.Infof("server %d get commit msg for block %d, from %d, current committed %d",
						self.Index, msgBlkNum, pMsg.Committer, self.GetCommittedBlockNo())
					self.timer.C <- &TimerEvent{
						evtType:  EventPeerHeartbeat,
						blockNum: pMsg.Committer,
					}
				}
				return
			}

			if isReady(self.getState()) {
				// start syncing check from proposed block round
				self.syncer.syncCheckReqC <- &SyncCheckReq{
					msg:      msg,
					blockNum: msgBlkNum,
				}
			}
		} else {
			// add to msg pool
			if err := self.msgPool.AddMsg(msg, msgHash); err != nil {
				log.Errorf("failed to add commit msg (%d) to pool", msgBlkNum)
				return
			}
			self.processConsensusMsg(msg)
		}
	case PeerHeartbeatMessage:
		pMsg, ok := msg.(*peerHeartbeatMsg)
		if !ok {
			log.Errorf("invalid msg with heartbeat msg type")
			return
		}
		self.processHeartbeatMsg(peerIdx, pMsg)
		if pMsg.CommittedBlockNumber+MAX_SYNCING_CHECK_BLK_NUM < self.GetCommittedBlockNo() {
			// delayed peer detected, response heartbeat with our chain Info
			self.timer.C <- &TimerEvent{
				evtType:  EventPeerHeartbeat,
				blockNum: peerIdx,
			}
		}

	case ProposalFetchMessage:
		pMsg, ok := msg.(*proposalFetchMsg)
		if !ok {
			log.Errorf("invalid msg with proposal fetch msg type")
			return
		}
		var pmsg *blockProposalMsg
		if self.Index == pMsg.ProposerID || pMsg.BlockNum == self.GetCurrentBlockNo() {
			pMsgs := self.msgPool.GetProposalMsgs(pMsg.BlockNum)
			for _, msg := range pMsgs {
				p := msg.(*blockProposalMsg)
				if p != nil && p.Block.getProposer() == pMsg.ProposerID {
					log.Infof("server %d rebroadcast proposal to %d, blk %d",
						self.Index, peerIdx, p.Block.getBlockNum())
					pmsg = p
				}
			}
		}
		if self.Index == pMsg.ProposerID {
			if pmsg == nil {
				blk, _ := self.blockPool.getSealedBlock(pMsg.BlockNum)
				if blk != nil {
					pmsg = &blockProposalMsg{
						Block: blk,
					}
				}
			}
		}

		if pmsg != nil {
			log.Infof("server %d, handle proposal fetch %d from %d",
				self.Index, pMsg.BlockNum, peerIdx)
			self.msgSendC <- &SendMsgEvent{
				ToPeer: peerIdx,
				Msg:    pmsg,
			}
		}

	case BlockFetchMessage:
		// handle block fetch msg
		pMsg, ok := msg.(*blockFetchMsg)
		if !ok {
			log.Errorf("invalid msg with blockfetch msg type")
			return
		}
		blk, blkHash := self.blockPool.getChainedBlock(pMsg.BlockNum)
		if blk == nil {
			return
		}
		msg := self.constructBlockFetchRespMsg(pMsg.BlockNum, blk, blkHash)
		log.Infof("server %d, handle blockfetch %d from %d",
			self.Index, pMsg.BlockNum, peerIdx)
		self.msgSendC <- &SendMsgEvent{
			ToPeer: peerIdx,
			Msg:    msg,
		}

	case BlockFetchRespMessage:
		self.syncer.syncMsgC <- &SyncMsg{
			fromPeer: peerIdx,
			msg:      msg,
		}

	case BlockInfoFetchMessage:
		// handle block Info fetch msg
		pMsg, ok := msg.(*BlockInfoFetchMsg)
		if !ok {
			log.Errorf("invalid msg with blockinfo fetch msg type")
			return
		}
		maxCnt := 64
		blkInfos := make([]*BlockInfo_, 0)
		targetBlkNum := self.GetCommittedBlockNo()
		for startBlkNum := pMsg.StartBlockNum; startBlkNum <= targetBlkNum; startBlkNum++ {
			blk, _ := self.blockPool.getSealedBlock(startBlkNum)
			if blk == nil {
				break
			}
			blkInfos = append(blkInfos, &BlockInfo_{
				BlockNum: startBlkNum,
				Proposer: blk.getProposer(),
			})
			if len(blkInfos) >= maxCnt {
				break
			}
		}
		msg := self.constructBlockInfoFetchRespMsg(blkInfos)
		log.Infof("server %d, response blockinfo fetch to %d, blk %d, len %d",
			self.Index, peerIdx, pMsg.StartBlockNum, len(blkInfos))
		self.msgSendC <- &SendMsgEvent{
			ToPeer: peerIdx,
			Msg:    msg,
		}

	case BlockInfoFetchRespMessage:
		self.syncer.syncMsgC <- &SyncMsg{
			fromPeer: peerIdx,
			msg:      msg,
		}
	case BlockSubmitMessage:
		pMsg, ok := msg.(*blockSubmitMsg)
		if !ok {
			log.Error("invalid msg with submit msg type")
			return
		}
		msgBlkNum := pMsg.GetBlockNum()
		if self.GetCurrentBlockNo() > msgBlkNum+1 {
			return
		}
		if err := self.msgPool.AddMsg(msg, msgHash); err != nil {
			if err != errDropFarFutureMsg {
				log.Errorf("failed to add submit msg (%d) to pool: %s", msgBlkNum, err)
			}
			return
		}
		if self.CheckSubmitBlock(msgBlkNum, pMsg.BlockStateRoot) {
			self.makeBlockSubmit(msgBlkNum)
		}
	}
}

func (self *Server) verifyCrossChainMsg(msg *blockProposalMsg) bool {
	root, err := self.blockPool.getCrossStatesRoot(msg.Block.Block.Header.Height - 1)
	if err != nil {
		log.Errorf("verifyCrossChainMsg:%s", err)
		return false
	}
	//malicious consensus node may create a nil cross chain msg proposal, but it is actual not nil.
	if root != common.UINT256_EMPTY && msg.Block.CrossChainMsg == nil {
		return false
	}
	if msg.Block.CrossChainMsg == nil {
		return true
	}
	if msg.Block.CrossChainMsg.StatesRoot != root ||
		msg.Block.CrossChainMsg.Version != types.CURR_CROSS_STATES_VERSION {
		return false
	}
	return true
}

func (self *Server) processProposalMsg(msg *blockProposalMsg) {
	msgBlkNum := msg.GetBlockNum()
	blk, prevBlkHash := self.blockPool.getSealedBlock(msg.GetBlockNum() - 1)
	if blk == nil {
		log.Errorf("BlockProposal failed to GetPreBlock:%d", msg.GetBlockNum()-1)
		return
	}

	msgPrevBlkHash := msg.Block.getPrevBlockHash()
	if prevBlkHash != msgPrevBlkHash {
		log.Errorf("BlockPrposalMessage check blocknum:%d,prevhash:%s,msg prevhash:%s", msg.GetBlockNum(), prevBlkHash.ToHexString(), msgPrevBlkHash.ToHexString())
		self.msgPool.DropMsg(msg)
		return
	}
	if self.LastConfigBlockNum != math.MaxUint32 && blk.Info.LastConfigBlockNum != self.LastConfigBlockNum {
		log.Errorf("BlockPrposalMessage  check LastConfigBlockNum blocknum:%d,prvLastConfigBlockNum:%d,self LastConfigBlockNum:%d", msg.GetBlockNum(), blk.Info.LastConfigBlockNum, self.LastConfigBlockNum)
		return
	}
	merkleRoot, err := self.blockPool.getExecMerkleRoot(msgBlkNum - 1)
	if err != nil {
		log.Errorf("failed to GetExecMerkleRoot: %s,blkNum:%d", err, msgBlkNum-1)
		return
	}
	if msg.Block.getPrevExecMerkleRoot() != merkleRoot {
		self.msgPool.DropMsg(msg)
		msgMerkleRoot := msg.Block.getPrevExecMerkleRoot()
		log.Errorf("BlockPrposalMessage check MerkleRoot blocknum:%d,msg MerkleRoot:%s,self MerkleRoot:%s", msg.GetBlockNum(), msgMerkleRoot.ToHexString(), merkleRoot.ToHexString())
		return
	}
	cfg := vconfig.ChainConfig{}
	if blk.getNewChainConfig() != nil {
		cfg = *blk.getNewChainConfig()
		chainCfg := self.GetChainConfig()
		if cfg.Hash() != chainCfg.Hash() {
			log.Errorf("processProposalMsg chainconfig unqeual to blockinfo cfg,view:(%d,%d),N:(%d,%d),C:(%d,%d),BlockMsgDelay:(%d,%d),HashMsgDelay:(%d,%d),PeerHandshakeTimeout:(%d,%d),posTable:(%v,%v),MaxBlockChangeView:(%d,%d)",
				cfg.View, chainCfg.View,
				cfg.N, chainCfg.N,
				cfg.C, chainCfg.C,
				cfg.BlockMsgDelay, chainCfg.BlockMsgDelay,
				cfg.HashMsgDelay, chainCfg.HashMsgDelay,
				cfg.PeerHandshakeTimeout, chainCfg.PeerHandshakeTimeout,
				cfg.PosTable, chainCfg.PosTable,
				cfg.MaxBlockChangeView, chainCfg.MaxBlockChangeView)
			self.msgPool.DropMsg(msg)
			return
		}
	}

	prevBlockTimestamp := blk.Block.Header.Timestamp
	currentBlockTimestamp := msg.Block.Block.Header.Timestamp
	if currentBlockTimestamp <= prevBlockTimestamp || currentBlockTimestamp > uint32(time.Now().Add(time.Minute*10).Unix()) {
		log.Errorf("BlockPrposalMessage check  blocknum:%d,prevBlockTimestamp:%d,currentBlockTimestamp:%d", msg.GetBlockNum(), prevBlockTimestamp, currentBlockTimestamp)
		self.msgPool.DropMsg(msg)
		return
	}

	// verify VRF
	proposerPk := self.peerPool.GetPeerPubKey(msg.Block.getProposer())
	if proposerPk == nil {
		log.Errorf("server %d failed to get proposer %d pk of block %d",
			self.Index, msg.Block.getProposer(), msgBlkNum)
		self.msgPool.DropMsg(msg)
		return
	}
	if err := verifyVrf(proposerPk, msgBlkNum, blk.getVrfValue(), msg.Block.getVrfValue(), msg.Block.getVrfProof()); err != nil {
		log.Errorf("server %d failed to verify vrf of block %d proposal from %d",
			self.Index, msgBlkNum, msg.Block.getProposer())
		self.msgPool.DropMsg(msg)
		return
	}
	if !self.verifyCrossChainMsg(msg) {
		log.Errorf("verify cross chain message error:%+v\n", msg.Block.CrossChainMsg)
		return
	}
	txs := msg.Block.Block.Transactions
	if len(txs) > 0 && self.nonSystxs(txs, msgBlkNum) {
		height := msgBlkNum - 1
		start, end := self.incrValidator.BlockRange()
		if msg.GetBlockNum() <= self.GetCompletedBlockNum() {
			log.Infof("processProposalMsg failed: MsgBlockNum:%d,CompletedBlockNum:%d", msg.GetBlockNum(), self.GetCompletedBlockNum())
			return
		}
		validHeight := height
		if height+1 == end {
			validHeight = start
		} else {
			self.incrValidator.Clean()
			//log.Infof("incr validator block height %v != ledger block height %v", int(end)-1, height)
			log.Infof("incr validator block height %v != ledger block height %v, CompletedBlockNum:%d", int(end)-1, height, self.GetCompletedBlockNum())
		}
		// start new routine to verify txs in proposal block
		go func() {
			if err := self.poolActor.VerifyBlock(txs, validHeight); err != nil && err != actor.ErrTimeout {
				log.Errorf("server %d verify proposal blk from %d failed, blk %d, txs %d, err: %s",
					self.Index, msg.Block.getProposer(), msgBlkNum, len(txs), err)
				return
			} else if err == actor.ErrTimeout {
				log.Errorf("server %d verify proposal blk from %d timedout, blk %d, txs %d, err: %s",
					self.Index, msg.Block.getProposer(), msgBlkNum, len(txs), err)
			}
			nonceCtx := make(map[common.Address]uint64)
			for _, tx := range txs {
				if err := self.incrValidator.Verify(tx, validHeight, nonceCtx); err != nil {
					log.Errorf("server %d verify proposal tx from %d failed, blk %d, txs %d, err: %s",
						self.Index, msg.Block.getProposer(), msgBlkNum, len(txs), err)
					return
				}
			}
			self.processConsensusMsg(msg)
		}()
	} else {
		// empty block, process directly
		self.processConsensusMsg(msg)
	}
}

func (self *Server) processConsensusMsg(msg ConsensusMsg) {
	if isReady(self.getState()) {
		self.msgC <- msg
	}
}

func (self *Server) processMsgEvent() error {
	select {
	case msg := <-self.msgC:

		log.Debugf("server %d process msg, block %d, type %d, current blk %d",
			self.Index, msg.GetBlockNum(), msg.Type(), self.GetCurrentBlockNo())

		switch msg.Type() {
		case BlockProposalMessage:
			pMsg := msg.(*blockProposalMsg)

			msgBlkNum := pMsg.GetBlockNum()
			if msgBlkNum == self.GetCurrentBlockNo() {
				// add proposal to block-pool
				if err := self.blockPool.newBlockProposal(pMsg); err != nil {
					// if err == errDupProposal {
					// 	// TODO: faulty proposer detected
					// }
					log.Errorf("failed to add block proposal (%d): %s", msgBlkNum, err)
					return nil
				}

				if self.isProposer(msgBlkNum, pMsg.Block.getProposer()) {
					// check if agreed on prev-blockhash
					if err := self.verifyPrevBlockHash(msgBlkNum, pMsg); err != nil {
						// continue
						log.Errorf("failed verify prevBlockHash from proposer %d, blk %d",
							pMsg.Block.getProposer(), msgBlkNum)
						return nil
					}

					// stop proposal timer
					self.timer.CancelProposalTimer(msgBlkNum)
					if self.isEndorser(msgBlkNum, self.Index) {
						if err := self.endorseBlock(pMsg, false); err != nil {
							log.Errorf("failed to endorse block proposal (%d): %s", msgBlkNum, err)
						}
					}
				} else {
					if self.isProposer(msgBlkNum, self.Index) {
						for _, msg := range self.msgPool.GetProposalMsgs(msgBlkNum) {
							p := msg.(*blockProposalMsg)
							if p != nil && p.Block.getProposer() == self.Index {
								log.Infof("server %d rebroadcast proposal to %d, blk %d",
									self.Index, pMsg.Block.getProposer(), msgBlkNum)
								self.broadcast(msg)
								break
							}
						}
					}
					// makeProposalTimeout handles non-leader proposals
				}
			} //  else {
			// 	// process new proposal when
			// 	// 1. we have endorsed for current BlockNum
			// 	// 2. proposal is from next potential-leader

			// 	// TODO
			// }

		case BlockEndorseMessage:
			pMsg := msg.(*blockEndorseMsg)
			msgBlkNum := pMsg.GetBlockNum()

			if msgBlkNum == self.GetCurrentBlockNo() {
				if pMsg.EndorsedProposer != self.Index && len(self.msgPool.GetProposalMsgs(msgBlkNum)) == 0 {
					self.fetchProposal(msgBlkNum, pMsg.EndorsedProposer)
				}

				// add endorse to block-pool
				self.blockPool.newBlockEndorsement(pMsg)
				log.Infof("server %d received endorse from %d, for proposer %d, block %d, empty: %t",
					self.Index, pMsg.Endorser, pMsg.EndorsedProposer, msgBlkNum, pMsg.EndorseForEmpty)

				// if had committed for current round, skip the following steps
				if self.blockPool.committedForBlock(msgBlkNum) {
					// get more endorse msg after committed, trigger seal-block-timeout
					if err := self.timer.StartCommitTimer(msgBlkNum); err != nil {
						log.Errorf("server %d start commit timer for %d(%d), block %d, err: %s",
							self.Index, pMsg.Endorser, pMsg.EndorsedProposer, msgBlkNum, err)
					}
					return nil
				}

				if self.isEndorser(msgBlkNum, pMsg.Endorser) {
					//              if countOfEndrosement(msg.proposal) >= 2C + 1:
					//                      stop WaitEndorsementTimer
					//                      commitBlock(msg.BlockHash)
					//              else if WaitEndorsementTimer has not started:
					//                      start WaitEndorsementTimer

					// TODO: should only count endorsements from endorsers
					if proposer, forEmpty, done := self.blockPool.endorseDone(msgBlkNum, self.GetChainConfig().C); done {
						// stop endorse timer
						self.timer.CancelEndorseMsgTimer(msgBlkNum)
						// stop empty endorse timer
						self.timer.CancelEndorseEmptyBlockTimer(msgBlkNum)
						proposal := self.findBlockProposal(msgBlkNum, proposer, forEmpty)
						if proposal == nil {
							log.Infof("server %d endorse %d done, waiting proposal from %d", self.Index, msgBlkNum, proposer)
						} else if self.isCommitter(msgBlkNum, self.Index) {
							// make endorsement
							if err := self.makeCommitment(proposal, msgBlkNum, forEmpty); err != nil {
								log.Errorf("failed to endorse for block %d: %s", msgBlkNum, err)
								return nil
							}
						}
					} // else {
					// 	// wait until endorse timeout
					// }
				} // else {
				// 	// makeEndorsementTimeout handles non-endorser endorsements
				// }
				if self.blockPool.endorseFailed(msgBlkNum, self.GetChainConfig().C) {
					// endorse failed, start empty endorsing
					self.timer.C <- &TimerEvent{
						evtType:  EventEndorseBlockTimeout,
						blockNum: msgBlkNum,
					}
				}
			} // else {
			// 	// process new endorsement when
			// 	// 1. we have committed for current BlockNum
			// 	// 2. endorsed proposal is from next potential-leader
			// }

		case BlockCommitMessage:
			pMsg := msg.(*blockCommitMsg)
			msgBlkNum := pMsg.GetBlockNum()

			if msgBlkNum == self.GetCurrentBlockNo() {
				//              if countOfCommitment(msg.proposal) >= 2C + 1:
				//                      stop WaitCommitsTimer
				//                      sealProposal(msg.BlockHash)
				//              else if WaitCommitsTimer has not started:
				//                      start WaitCommitsTimer
				if err := self.blockPool.newBlockCommitment(pMsg); err != nil {
					log.Errorf("failed to add commit msg (%d): %s", msgBlkNum, err)
					return nil
				}

				log.Infof("server %d received commit from %d, for proposer %d, block %d, empty: %t",
					self.Index, pMsg.Committer, pMsg.BlockProposer, msgBlkNum, pMsg.CommitForEmpty)

				chainCfg := self.GetChainConfig()
				if proposer, forEmpty, done := self.blockPool.commitDone(msgBlkNum, chainCfg.C, chainCfg.N); done {
					self.blockPool.setCommitDone(msgBlkNum)
					proposal := self.findBlockProposal(msgBlkNum, proposer, forEmpty)
					if proposal == nil {
						// TODO: commit done, but we not have the proposal, should request proposal from neighbours
						//       commitTimeout handle this
						log.Infof("server %d commit %d done, waiting proposal",
							self.Index, msgBlkNum)
						return nil
					}

					if self.isCommitter(msgBlkNum, self.Index) {
						// make sure committer broadcasting his commit msg
						if err := self.makeCommitment(proposal, msgBlkNum, forEmpty); err != nil {
							log.Errorf("server %d consensused %d, committer broadcast commit msg: %s", self.Index, msgBlkNum, err)
						}
					}

					// stop commit timer
					self.timer.CancelCommitMsgTimer(msgBlkNum)

					if err := self.makeSealed(proposal, forEmpty); err != nil {
						log.Errorf("failed to seal block %d, err: %s", msgBlkNum, err)
					}
				} // else {
				// 	// wait commit timeout, nothing to do
				// }

			} //  else {
			// 	// nothing to do besides adding to msg pool

			// 	// FIXME: add msg from msg-pool to block-pool when starting new block-round
			// }
		}

	case <-self.quitC:
		return fmt.Errorf("server %d, processMsgEvent loop quit", self.Index)
	}
	return nil
}

func (self *Server) actionLoop() {
	self.quitWg.Add(1)
	defer self.quitWg.Done()

	for {
		select {
		case action := <-self.bftActionC:
			switch action.Type {
			case MakeProposal:
				// this may triggered when block sealed or random backoff of 2nd proposer
				blkNum := self.GetCurrentBlockNo()
				if blkNum > action.BlockNum {
					continue
				}

				var proposal *blockProposalMsg
				msgs := self.msgPool.GetProposalMsgs(blkNum)
				for _, m := range msgs {
					if p, ok := m.(*blockProposalMsg); ok && p.Block.getProposer() == self.Index {
						proposal = p
						break
					}
				}
				if proposal == nil {
					if err := self.makeProposal(blkNum, action.forEmpty); err != nil {
						log.Errorf("server %d failed to making proposal (%d): %s",
							self.Index, blkNum, err)
					}
				}

			case EndorseBlock:
				// endorse the proposal
				blkNum := action.Proposal.GetBlockNum()
				if err := self.endorseBlock(action.Proposal, action.forEmpty); err != nil {
					log.Errorf("server %d failed to endorse block proposal (%d): %s",
						self.Index, blkNum, err)
					continue
				}

			case CommitBlock:
				blkNum := action.Proposal.GetBlockNum()
				if err := self.commitBlock(action.Proposal, action.forEmpty); err != nil {
					log.Errorf("server %d failed to commit block proposal (%d): %s",
						self.Index, blkNum, err)
					continue
				}
			case SealBlock:
				if action.Proposal.GetBlockNum() < self.GetCurrentBlockNo() {
					continue
				}
				if err := self.sealProposal(action.Proposal, action.forEmpty); err != nil {
					log.Errorf("server %d failed to seal block (%d): %s",
						self.Index, action.Proposal.GetBlockNum(), err)
				}
			case FastForward:
				// 1. from current block num, check commit msgs in msg pool
				// 2. if commit consensused, seal the proposal
				for {
					blkNum := self.GetCurrentBlockNo()
					chainCfg := self.GetChainConfig()

					if err := self.updateParticipantConfig(); err != nil {
						log.Errorf("server %d update config failed in forwarding: %s", self.Index, err)
					}

					// get pending msgs from msgpool
					pMsgs := self.msgPool.GetProposalMsgs(blkNum)
					for _, msg := range pMsgs {
						p := msg.(*blockProposalMsg)
						if p != nil {
							if err := self.blockPool.newBlockProposal(p); err != nil {
								log.Errorf("server %d failed add proposal in fastforwarding: %s",
									self.Index, err)
							}
						}
					}

					cMsgs := self.msgPool.GetCommitMsgs(blkNum)
					commitMsgs := make([]*blockCommitMsg, 0)
					for _, msg := range cMsgs {
						c := msg.(*blockCommitMsg)
						if c != nil {
							if err := self.blockPool.newBlockCommitment(c); err == nil {
								commitMsgs = append(commitMsgs, c)
							} else {
								log.Errorf("server %d failed to add commit in fastforwarding: %s",
									self.Index, err)
							}
						}
					}

					log.Infof("server %d fastforwarding from %d, (%d, %d)",
						self.Index, self.GetCurrentBlockNo(), len(cMsgs), len(pMsgs))
					if len(pMsgs) == 0 && len(cMsgs) == 0 {
						log.Infof("server %d fastforward done, no msg", self.Index)
						self.startNewRound()
						break
					}

					// check if consensused
					proposer, forEmpty := getCommitConsensus(commitMsgs, int(chainCfg.C), int(chainCfg.N))
					if proposer == math.MaxUint32 {
						if err := self.catchConsensus(blkNum); err != nil {
							log.Infof("server %d fastforward done, catch consensus: %s", self.Index, err)
						}
						log.Infof("server %d fastforward done at blk %d, no consensus", self.Index, blkNum)
						break
					}

					// get consensused proposal
					var proposal *blockProposalMsg
					for _, m := range pMsgs {
						p, ok := m.(*blockProposalMsg)
						if !ok {
							continue
						}
						if p.Block.getProposer() == proposer {
							proposal = p
							break
						}
					}
					if proposal == nil {
						log.Infof("server %d fastforward stopped at blk %d, no proposal", self.Index, blkNum)
						self.fetchProposal(blkNum, proposer)
						if err := self.timer.StartCommitTimer(blkNum); err != nil {
							log.Errorf("server %d fastforward startcommittimer at blk %d, err:%s", self.Index, blkNum, err)
						}
						break
					}

					log.Infof("server %d fastforwarding block %d, proposer %d",
						self.Index, blkNum, proposal.Block.getProposer())

					// fastforward the block
					if err := self.sealBlock(proposal.Block, forEmpty, true); err != nil {
						log.Errorf("server %d fastforward stopped at blk %d, seal failed: %s",
							self.Index, blkNum, err)
						break
					}
				}

			case ReBroadcast:
				blkNum := self.GetCurrentBlockNo()
				if blkNum > action.BlockNum {
					continue
				}

				proposals := make([]*blockProposalMsg, 0)
				for _, msg := range self.msgPool.GetProposalMsgs(blkNum) {
					p := msg.(*blockProposalMsg)
					if p != nil {
						proposals = append(proposals, p)
					}
				}

				for _, p := range proposals {
					if p.Block.getProposer() == self.Index {
						log.Infof("server %d rebroadcast proposal, blk %d",
							self.Index, p.Block.getBlockNum())
						self.broadcast(p)
					}
				}
				if self.isEndorser(blkNum, self.Index) {
					rebroadcasted := false
					endorseFailed := self.blockPool.endorseFailed(blkNum, self.GetChainConfig().C)
					eMsgs := self.msgPool.GetEndorsementsMsgs(blkNum)
					for _, msg := range eMsgs {
						e := msg.(*blockEndorseMsg)
						if e != nil && e.Endorser == self.Index && e.EndorseForEmpty == endorseFailed {
							log.Infof("server %d rebroadcast endorse, blk %d for %d, %t",
								self.Index, e.GetBlockNum(), e.EndorsedProposer, e.EndorseForEmpty)
							self.broadcast(e)
							rebroadcasted = true
						}
					}
					if !rebroadcasted {
						proposal := self.getHighestRankProposal(blkNum, proposals)
						if proposal != nil {
							if err := self.endorseBlock(proposal, false); err != nil {
								log.Errorf("server %d rebroadcasting failed to endorse (%d): %s",
									self.Index, blkNum, err)
							}
						} else {
							log.Errorf("server %d rebroadcasting failed to endorse(%d), no proposal found(%d)",
								self.Index, blkNum, len(proposals))
						}
					}
				} else if proposal, forEmpty := self.blockPool.getEndorsedProposal(blkNum); proposal != nil {
					// construct endorse msg
					if endorseMsg, _ := self.constructEndorseMsg(proposal, forEmpty); endorseMsg != nil {
						self.broadcast(endorseMsg)
					}
				}
				if self.isCommitter(blkNum, self.Index) {
					committed := false
					cMsgs := self.msgPool.GetCommitMsgs(self.GetCurrentBlockNo())
					for _, msg := range cMsgs {
						c := msg.(*blockCommitMsg)
						if c != nil && c.Committer == self.Index {
							log.Infof("server %d rebroadcast commit, blk %d for %d, %t",
								self.Index, c.GetBlockNum(), c.BlockProposer, c.CommitForEmpty)
							self.broadcast(msg)
							committed = true
						}
					}
					if !committed {
						if proposer, forEmpty, done := self.blockPool.endorseDone(blkNum, self.GetChainConfig().C); done {
							proposal := self.findBlockProposal(blkNum, proposer, forEmpty)

							// consensus ok, make endorsement
							if proposal == nil {
								self.fetchProposal(blkNum, proposer)
								// restart endorsing timer
								if err := self.timer.StartEndorsingTimer(blkNum); err != nil {
									log.Errorf("server %d endorse %d done, startendorsingtimer err:%s", self.Index, blkNum, err)
								}
								log.Errorf("server %d endorse %d done, but no proposal", self.Index, blkNum)
							} else if err := self.makeCommitment(proposal, blkNum, forEmpty); err != nil {
								log.Errorf("server %d failed to commit block %d on rebroadcasting: %s",
									self.Index, blkNum, err)
							}
						} else if self.blockPool.endorseFailed(blkNum, self.GetChainConfig().C) {
							// endorse failed, start empty endorsing
							self.timer.C <- &TimerEvent{
								evtType:  EventEndorseBlockTimeout,
								blockNum: blkNum,
							}
						}
					}
				}
			case SubmitBlock:
				blkNum := self.GetCurrentBlockNo()
				if action.BlockNum > blkNum {
					continue
				}
				stateRoot, err := self.blockPool.getExecMerkleRoot(action.BlockNum)
				if err != nil {
					log.Infof("handleBlockSubmit failed:%s", err)
					continue
				}
				if self.CheckSubmitBlock(action.BlockNum, stateRoot) {
					if err := self.blockPool.submitBlock(action.BlockNum); err != nil {
						log.Errorf("SubmitBlock err:%s", err)
					}
				}
			}
		case <-self.quitC:
			log.Infof("server %d actionLoop quit", self.Index)
			return
		}
	}
}

func (self *Server) timerLoop() {
	self.quitWg.Add(1)
	defer self.quitWg.Done()

	for {
		select {
		case evt := <-self.timer.C:
			if err := self.processTimerEvent(evt); err != nil {
				log.Errorf("failed to process timer evt: %d, err: %s", evt.evtType, err)
			}

		case <-self.quitC:
			log.Infof("server %d timerLoop quit", self.Index)
			return
		}
	}
}

func (self *Server) processTimerEvent(evt *TimerEvent) error {
	switch evt.evtType {
	case EventProposalBackoff:
		// 1. if endorsed, return
		// 2. if no proposal received,
		// 		if 2nd proposer, make proposal, start endorse timeout, return
		//		else, return (wait proposal timeout)
		// 3. else:
		// 		if no proposal from leader, return (wait proposal timeout will make endorse anyway)
		// 		else, return (endorsing on leader-proposal done when received the proposal)
		//

		if self.blockPool.endorsedForBlock(evt.blockNum) {
			return nil
		}
		if !isReady(self.getState()) {
			return nil
		}
		proposals := self.blockPool.getBlockProposals(evt.blockNum)
		if len(proposals) == 0 {
			// no proposal received, make proposal, start endorse timeout
			if self.is2ndProposer(evt.blockNum, self.Index) {
				if err := self.makeProposal(evt.blockNum, false); err != nil {
					return fmt.Errorf("failed to make 2nd proposal (%d): %s", evt.blockNum, err)
				}
			}
		}

	case EventProposeBlockTimeout:
		// 1. if endorsed, return
		// 2. check proposal from leader, if there is, endorse the proposal, start endorse timeout, return
		// 3. check proposal from 2nd proposer, endorse for first 2nd proposal, start endorse timeout, return
		// 4. if not, random backoff, return
		//
		// then propose for empty block, start 2ndProposal timeout, return
		//
		return self.handleProposalTimeout(evt)

	case EventRandomBackoff:
		// 1. if endorsed, return
		// 2. if any valid proposal, endorse on high-priority one (priority from vrf), start endorse timeout, return
		// 3. make empty proposal, broadcast, start 2nd proposal timeout, return
		//
		return self.handleProposalTimeout(evt)

	case EventPropose2ndBlockTimeout:
		// 1. if endorsed, return
		// 2. there must some valid proposal, if not, force resync, reset peer neighbours
		// 3. endorse on highest-priority one, start endorse timeout, return
		//
		return self.handleProposalTimeout(evt)

	case EventEndorseBlockTimeout:
		// 1. if committed, return
		// 2. check endorse quorum
		// 3. if quorum reached, endorse the proposal, start commit timeout, return
		// 4. broadcast endorse on highest-priority proposal empty, start empty endorse timeout, return
		//
		if self.blockPool.committedForBlock(evt.blockNum) {
			return nil
		}
		if !isReady(self.getState()) {
			return nil
		}
		if proposer, forEmpty, done := self.blockPool.endorseDone(evt.blockNum, self.GetChainConfig().C); done {
			proposal := self.findBlockProposal(evt.blockNum, proposer, forEmpty)

			// consensus ok, make endorsement
			if proposal == nil {
				self.fetchProposal(evt.blockNum, proposer)
				// restart endorsing timer
				if err := self.timer.StartEndorsingTimer(evt.blockNum); err != nil {
					log.Errorf("endorse %d done,fetchproposal  startendorsingtimer err:%s", evt.blockNum, err)
				}
				return fmt.Errorf("endorse %d done, but no proposal available", evt.blockNum)
			}
			if err := self.verifyPrevBlockHash(evt.blockNum, proposal); err != nil {
				// restart endorsing timer
				if errinfo := self.timer.StartEndorsingTimer(evt.blockNum); errinfo != nil {
					log.Errorf("endorse %d done,verifyprevblockhash, startendorsingtimer err:%s", evt.blockNum, errinfo)
				}
				return fmt.Errorf("endorse %d done, but prev blk hash inconsistency: %s", evt.blockNum, err)
			}
			if err := self.makeCommitment(proposal, evt.blockNum, forEmpty); err != nil {
				return fmt.Errorf("failed to endorse for block %d on endorse timeout: %s", evt.blockNum, err)
			}
			return nil
		}
		if !isActive(self.getState()) {
			// not active yet, waiting active peers making decision
			return nil
		}
		if self.blockPool.endorsedForEmptyBlock(evt.blockNum) {
			return nil
		}
		proposals := self.blockPool.getBlockProposals(evt.blockNum)
		if len(proposals) == 0 {
			log.Errorf("endorsing timeout, without any proposal. restarting syncing")
			self.restartSyncing()
			return nil
		}
		proposal := self.getHighestRankProposal(evt.blockNum, proposals)
		if proposal != nil {
			if err := self.endorseBlock(proposal, true); err != nil {
				return fmt.Errorf("failed to endorse block proposal (%d): %s", evt.blockNum, err)
			}
		}
		return nil

	case EventEndorseEmptyBlockTimeout:
		// 1. if committed, return
		// 2. check endorse quorum
		// 3. if quorum reached, commit the proposal, start commit timeout, return
		// 4. check empty endorse quorum
		// 5. if empty endorse quorum reached, commit the empty proposal, start commit timeout, return
		//
		if self.blockPool.committedForBlock(evt.blockNum) {
			return nil
		}
		if !isReady(self.getState()) {
			return nil
		}
		if proposer, forEmpty, done := self.blockPool.endorseDone(evt.blockNum, self.GetChainConfig().C); done {
			proposal := self.findBlockProposal(evt.blockNum, proposer, forEmpty)

			// consensus ok, make endorsement
			if proposal == nil {
				self.fetchProposal(evt.blockNum, proposer)
				// restart timer
				if err := self.timer.StartEndorseEmptyBlockTimer(evt.blockNum); err != nil {
					return fmt.Errorf("failed to startendorseemptyblocktimer block %d err:%s", evt.blockNum, err)
				}
			} else if err := self.makeCommitment(proposal, evt.blockNum, forEmpty); err != nil {
				return fmt.Errorf("failed to endorse for block %d on empty endorse timeout: %s", evt.blockNum, err)
			}
			return nil
		} else {
			log.Errorf("server %d: empty endorse timeout, no quorum", self.Index)
			if !isActive(self.getState()) {
				proposals := self.blockPool.getBlockProposals(evt.blockNum)
				proposal := self.getHighestRankProposal(evt.blockNum, proposals)
				if proposal != nil {
					if err := self.endorseBlock(proposal, true); err != nil {
						return fmt.Errorf("failed to endorse block proposal (%d): %s", evt.blockNum, err)
					}
				}
			} else {
				if err := self.timer.StartEndorseEmptyBlockTimer(evt.blockNum); err != nil {
					return fmt.Errorf("failed to endorse block proposal (%d): StartEndorseEmptyBlockTimer err: %s", evt.blockNum, err)
				}
			}
		}
		return nil

	case EventCommitBlockTimeout:
		// 1. if sealed, return
		// 2. check commit quorum
		// 3. if quorum reached, seal the commit, start new round, return
		// 4. else: there must have some network issues, force resync, reset all neighbours
		//
		if blk, _ := self.blockPool.getSealedBlock(evt.blockNum); blk != nil {
			return nil
		}
		if !isReady(self.getState()) {
			return nil
		}
		if !self.blockPool.isCommitHadDone(evt.blockNum) {
			chainCfg := self.GetChainConfig()
			if proposer, forEmpty, done := self.blockPool.commitDone(evt.blockNum, chainCfg.C, chainCfg.N); done {
				self.blockPool.setCommitDone(evt.blockNum)
				proposal := self.findBlockProposal(evt.blockNum, proposer, forEmpty)
				if proposal == nil {
					self.restartSyncing()
					return fmt.Errorf("commit timeout, consensused proposal not available. need resync")
				}

				if err := self.makeSealed(proposal, forEmpty); err != nil {
					return fmt.Errorf("commit timeout, failed to seal block %d: %s", evt.blockNum, err)
				}
				return nil
			} else {
				log.Errorf("server %d commit blk %d timeout without consensus", self.Index, evt.blockNum)
				self.restartSyncing()
			}
		}

	case EventPeerHeartbeat:
		self.heartbeat()

	case EventTxPool:
		self.timer.stopTxTicker(evt.blockNum)
		if self.GetCompletedBlockNum()+1 == evt.blockNum {
			validHeight := self.validHeight(evt.blockNum)
			newProposal := false
			nonceCtx := make(map[common.Address]uint64)
			for _, e := range self.poolActor.GetTxnPool(true, validHeight) {
				if err := self.incrValidator.Verify(e.Tx, validHeight, nonceCtx); err == nil {
					newProposal = true
					break
				}
			}
			if newProposal {
				self.timer.CancelTxBlockTimeout(evt.blockNum)
				self.startNewProposal(evt.blockNum)
			} else {
				//reset timer, continue waiting txs from txnpool
				self.timer.startTxTicker(evt.blockNum)
			}
		} else {
			self.timer.startTxTicker(evt.blockNum)
		}
	case EventTxBlockTimeout:
		self.timer.stopTxTicker(evt.blockNum)
		self.timer.CancelTxBlockTimeout(evt.blockNum)
		self.startNewProposal(evt.blockNum)
	}
	return nil
}

func (self *Server) processHandshakeMsg(peerIdx uint32, msg *peerHandshakeMsg) error {
	self.peerPool.peerHandshake(peerIdx, msg)
	self.stateMgr.StateEventC <- &StateEvent{
		Type: UpdatePeerConfig,
		peerState: &PeerState{
			peerIdx:           peerIdx,
			connected:         true,
			chainConfigView:   msg.ChainConfig.View,
			committedBlockNum: msg.CommittedBlockNumber,
		},
	}

	return nil
}

func (self *Server) processHeartbeatMsg(peerIdx uint32, msg *peerHeartbeatMsg) {
	self.peerPool.peerHeartbeat(peerIdx, msg)
	log.Debugf("server %d received heartbeat from peer %d, chainview %d, blkNum %d",
		self.Index, peerIdx, msg.ChainConfigView, msg.CommittedBlockNumber)
	self.stateMgr.StateEventC <- &StateEvent{
		Type: UpdatePeerState,
		peerState: &PeerState{
			peerIdx:           peerIdx,
			connected:         true,
			chainConfigView:   msg.ChainConfigView,
			committedBlockNum: msg.CommittedBlockNumber,
		},
	}
}

func (self *Server) endorseBlock(proposal *blockProposalMsg, forEmpty bool) error {
	// for each round, one node can only endorse one block, or empty block
	if proposal.Block.getProposer() == self.Index {
		return nil
	}
	blkNum := proposal.GetBlockNum()

	// check if has endorsed
	if !forEmpty && self.blockPool.endorsedForBlock(blkNum) {
		return nil
	} else if forEmpty && self.blockPool.endorsedForEmptyBlock(blkNum) {
		return nil
	}

	if !forEmpty {
		if self.blockPool.endorseFailed(blkNum, self.GetChainConfig().C) {
			forEmpty = true
			log.Errorf("server %d, endorsing %d, changed from true to false", self.Index, blkNum)
		}
	}

	// build endorsement msg
	endorseMsg, err := self.constructEndorseMsg(proposal, forEmpty)
	if err != nil {
		return fmt.Errorf("failed to construct endorse msg: %s", err)
	}

	// set the block as self-endorsed-block
	if err := self.blockPool.setProposalEndorsed(proposal, forEmpty); err != nil {
		return fmt.Errorf("failed to set proposal as endorsed: %s", err)
	}

	self.processConsensusMsg(endorseMsg)
	// if node is endorser of current round
	if forEmpty || self.isEndorser(blkNum, self.Index) {
		h, _ := HashMsg(endorseMsg)
		self.msgPool.AddMsg(endorseMsg, h)
		log.Infof("endorser %d, endorsed block %d, from server %d",
			self.Index, blkNum, proposal.Block.getProposer())
		// broadcast my endorsement
		self.broadcast(endorseMsg)
		return nil
	}

	// start endorsing timer
	// TODO: endorsing may have reached consensus before received proposal, handle this
	if !forEmpty {
		if err := self.timer.StartEndorsingTimer(blkNum); err != nil {
			log.Errorf("endorseblock blk:%d,startendorsingtimer err:%s", blkNum, err)
			return fmt.Errorf("endorseblock blk:%d,startendorsingkimer err:%s", blkNum, err)
		}
	} else {
		if err := self.timer.StartEndorseEmptyBlockTimer(blkNum); err != nil {
			return fmt.Errorf("endorseBlock blk:%d,forEmpty:%v,StartEndorsingTimer err:%s", blkNum, forEmpty, err)
		}
	}

	return nil
}

func (self *Server) commitBlock(proposal *blockProposalMsg, forEmpty bool) error {
	// for each round, we can only commit one block
	if proposal.Block.getProposer() == self.Index {
		return nil
	}
	blkNum := proposal.GetBlockNum()
	if self.blockPool.committedForBlock(blkNum) {
		return nil
	}

	var blkHash common.Uint256
	if !forEmpty {
		blkHash = proposal.Block.Block.Hash()
	} else {
		if proposal.Block.EmptyBlock == nil {
			return fmt.Errorf("blk %d proposal from %d has no empty proposal", blkNum, proposal.Block.getProposer())
		}
		blkHash = proposal.Block.EmptyBlock.Hash()
	}

	endorses := make([]*blockEndorseMsg, 0)
	for _, msg := range self.msgPool.GetEndorsementsMsgs(blkNum) {
		if e := msg.(*blockEndorseMsg); e != nil {
			if bytes.Compare(blkHash[:], e.EndorsedBlockHash[:]) == 0 && e.EndorseForEmpty == forEmpty {
				endorses = append(endorses, e)
			}
		}
	}

	// build commit msg
	commitMsg, err := self.constructCommitMsg(proposal, endorses, forEmpty)
	if err != nil {
		return fmt.Errorf("failed to construct commit msg: %s", err)
	}

	// set the block as committed-block
	if err := self.blockPool.setProposalCommitted(proposal, forEmpty); err != nil {
		return fmt.Errorf("failed to set proposal as committed: %s", err)
	}

	self.processConsensusMsg(commitMsg)
	// if node is committer of current round
	if forEmpty || self.isCommitter(blkNum, self.Index) {
		h, _ := HashMsg(commitMsg)
		self.msgPool.AddMsg(commitMsg, h)
		log.Infof("committer %d, set block %d committed, from server %d",
			self.Index, blkNum, proposal.Block.getProposer())
		// broadcast my commitment
		self.broadcast(commitMsg)
		return nil
	}

	// start commit timer
	// TODO: committing may have reached consensus before received endorsement, handle this
	if err := self.timer.StartCommitTimer(blkNum); err != nil {
		return fmt.Errorf("commitBlock startcommittimer err: %s", err)
	}
	return nil
}

//
// Note: sealProposal updates self.currentBlockNum, make sure not concurrency
// (only called by sealProposal action)
//
func (self *Server) sealProposal(proposal *blockProposalMsg, empty bool) error {
	// for each round, we can only seal one block
	if err := self.sealBlock(proposal.Block, empty, true); err != nil {
		return err
	}

	if self.hasBlockConsensused() {
		self.makeFastForward()
	} else {
		return self.startNewRound()
	}

	return nil
}

func (self *Server) fastForwardBlock(block *Block) error {

	// TODO: update chainconfig when forwarding

	if isActive(self.getState()) {
		return fmt.Errorf("server %d: invalid fastforward, current state: %d", self.Index, self.getState())
	}
	if self.GetCurrentBlockNo() > block.getBlockNum() {
		return nil
	}
	if self.GetCurrentBlockNo() == block.getBlockNum() {
		// block from peer syncer, there should only one candidate block
		flag := false
		if len(block.Block.Header.SigData) <= 1 {
			flag = true
		}
		return self.sealBlock(block, false, flag)
	}
	return fmt.Errorf("server %d: fastforward blk %d failed, current blkNum: %d",
		self.Index, block.getBlockNum(), self.GetCurrentBlockNo())
}

func (self *Server) sealBlock(block *Block, empty bool, sigdata bool) error {
	sealedBlkNum := block.getBlockNum()
	if sealedBlkNum < self.GetCurrentBlockNo() {
		// we already in future round
		log.Errorf("late seal of %d, current blkNum: %d", sealedBlkNum, self.GetCurrentBlockNo())
		return nil
	} else if sealedBlkNum > self.GetCurrentBlockNo() {
		// we have lost sync, restarting syncing
		self.restartSyncing()
		return fmt.Errorf("future seal of %d, current blknum: %d", sealedBlkNum, self.GetCurrentBlockNo())
	}

	if err := self.blockPool.setBlockSealed(block, empty, sigdata); err != nil {
		return fmt.Errorf("failed to seal proposal: %s", err)
	}

	// TODO: also persistent the block endorsers and committer msgs

	// notify other modules that block sealed
	self.timer.onBlockSealed(sealedBlkNum)
	self.msgPool.onBlockSealed(sealedBlkNum)
	self.blockPool.onBlockSealed(sealedBlkNum)

	_, h := self.blockPool.getSealedBlock(sealedBlkNum)
	prevBlkHash := block.getPrevBlockHash()
	log.Infof("server %d, sealed block %d, proposer %d, prevhash: %s, hash: %s", self.Index,
		sealedBlkNum, block.getProposer(), prevBlkHash.ToHexString(), h.ToHexString())

	// broadcast to other modules
	// TODO: block committed, update tx pool, notify block-listeners
	{
		if sealedBlkNum >= self.GetCurrentBlockNo() {
			self.SetCurrentBlockNo(sealedBlkNum + 1)
		}
	}
	return nil
}

func (self *Server) msgSendLoop() {
	self.quitWg.Add(1)
	defer self.quitWg.Done()

	for {
		select {
		case evt := <-self.msgSendC:
			if self.nonConsensusNode() {
				continue
			}
			payload, err := SerializeVbftMsg(evt.Msg)
			if err != nil {
				log.Errorf("server %d failed to serialized msg (type: %d): %s", self.Index, evt.Msg.Type(), err)
				continue
			}
			if evt.ToPeer == math.MaxUint32 {
				// broadcast
				self.broadcastToAll(payload)
			} else {
				if err := self.sendToPeer(evt.ToPeer, payload); err != nil {
					log.Errorf("server %d xmit to peer %d failed: %s", self.Index, evt.ToPeer, err)
				}
			}

		case <-self.quitC:
			log.Infof("server %d msgSendLoop quit", self.Index)
			return
		}
	}
}

//creategovernaceTransaction invoke governance native contract commit_pos
func (self *Server) creategovernaceTransaction(blkNum uint32) (*types.Transaction, error) {
	mutable := utils.BuildNativeTransaction(nutils.GovernanceContractAddress, gover.COMMIT_DPOS, []byte{})
	mutable.Nonce = blkNum
	tx, err := mutable.IntoImmutable()
	return tx, err
}

//checkNeedUpdateChainConfig use blockcount
func (self *Server) checkNeedUpdateChainConfig(blockNum uint32) bool {
	prevBlk, _ := self.blockPool.getSealedBlock(blockNum - 1)
	if prevBlk == nil {
		log.Errorf("failed to get prevBlock (%d)", blockNum-1)
		return false
	}
	lastConfigBlkNum := prevBlk.getLastConfigBlockNum()
	if (blockNum - lastConfigBlkNum) >= self.GetChainConfig().MaxBlockChangeView {
		return true
	}
	return false
}

//checkUpdateChainConfig query leveldb check is force update
func (self *Server) checkUpdateChainConfig(blkNum uint32) bool {
	force, err := isUpdate(self.blockPool.getExecWriteSet(blkNum-1), self.GetChainConfig().View)
	if err != nil {
		log.Errorf("checkUpdateChainConfig err:%s", err)
		return false
	}
	log.Debugf("checkUpdateChainConfig force: %v", force)
	return force
}

func (self *Server) validHeight(blkNum uint32) uint32 {
	height := blkNum - 1
	validHeight := height
	start, end := self.incrValidator.BlockRange()
	if height+1 == end {
		validHeight = start
	} else {
		self.incrValidator.Clean()
	}
	return validHeight
}

func (self *Server) nonSystxs(sysTxs []*types.Transaction, blkNum uint32) bool {
	if self.checkNeedUpdateChainConfig(blkNum) && len(sysTxs) == 1 {
		invoke := sysTxs[0].Payload.(*payload.InvokeCode)
		if invoke == nil {
			log.Errorf("nonSystxs invoke is nil,blocknum:%d", blkNum)
			return true
		}
		if bytes.Compare(invoke.Code, ninit.COMMIT_DPOS_BYTES) == 0 {
			return false
		}
	}
	return true
}

func (self *Server) makeProposal(blkNum uint32, forEmpty bool) error {
	if blkNum < self.GetCurrentBlockNo() {
		return fmt.Errorf("server %d ignore deprecatd blk proposal %d, current %d",
			self.Index, blkNum, self.GetCurrentBlockNo())
	}

	validHeight := self.validHeight(blkNum)
	sysTxs := make([]*types.Transaction, 0)
	userTxs := make([]*types.Transaction, 0)

	//check need upate chainconfig
	var cfg *vconfig.ChainConfig
	if self.checkNeedUpdateChainConfig(blkNum) || self.checkUpdateChainConfig(blkNum) {
		chainconfig, err := getChainConfig(self.blockPool.getExecWriteSet(blkNum-1), blkNum)
		if err != nil {
			return fmt.Errorf("getChainConfig failed:%s", err)
		}
		//add transaction invoke governance native commit_pos contract
		if self.checkNeedUpdateChainConfig(blkNum) {
			tx, err := self.creategovernaceTransaction(blkNum)
			if err != nil {
				return fmt.Errorf("construct governace transaction error: %v", err)
			}
			sysTxs = append(sysTxs, tx)
			chainconfig.View++
		}
		forEmpty = true
		cfg = chainconfig
	}
	if self.nonConsensusNode() {
		return fmt.Errorf("%d quit consensus node", self.Index)
	}

	if !forEmpty {
		nonceCtx := make(map[common.Address]uint64)
		for _, e := range self.poolActor.GetTxnPool(true, validHeight) {
			if err := self.incrValidator.Verify(e.Tx, validHeight, nonceCtx); err == nil {
				userTxs = append(userTxs, e.Tx)
			}
		}
		log.Infof("make proposal get %d valid tx from pool", len(userTxs))
	}

	proposal, err := self.constructProposalMsg(blkNum, sysTxs, userTxs, cfg)
	if err != nil {
		return fmt.Errorf("failed to construct proposal: %s", err)
	}

	log.Infof("server %d make proposal for block %d", self.Index, blkNum)

	// add proposal to self
	h, _ := HashMsg(proposal)
	self.msgPool.AddMsg(proposal, h)
	self.processProposalMsg(proposal)
	self.broadcast(proposal)
	return nil
}

func (self *Server) makeCommitment(proposal *blockProposalMsg, blkNum uint32, forEmpty bool) error {
	if err := self.commitBlock(proposal, forEmpty); err != nil {
		return fmt.Errorf("failed to commit block proposal (%d): %s", blkNum, err)
	}
	return nil
}

func (self *Server) makeSealed(proposal *blockProposalMsg, forEmpty bool) error {
	blkNum := proposal.GetBlockNum()

	if err := self.verifyPrevBlockHash(blkNum, proposal); err != nil {
		// TODO: in-consistency with prev-blockhash, resync-required
		self.restartSyncing()
		return fmt.Errorf("verify prev block hash failed: %s", err)
	}

	log.Infof("server %d ready to seal block %d, for proposer %d, empty: %t",
		self.Index, blkNum, proposal.Block.getProposer(), forEmpty)

	// seal the block
	self.bftActionC <- &BftAction{
		Type:     SealBlock,
		BlockNum: blkNum,
		Proposal: proposal,
		forEmpty: forEmpty,
	}
	return nil
}

func (self *Server) makeFastForward() {
	go func() {
		self.bftActionC <- &BftAction{
			Type:     FastForward,
			BlockNum: self.GetCurrentBlockNo(),
		}
	}()
}

func (self *Server) reBroadcastCurrentRoundMsgs() {
	go func() {
		self.bftActionC <- &BftAction{
			Type:     ReBroadcast,
			BlockNum: self.GetCurrentBlockNo(),
		}
	}()
}

func (self *Server) makeBlockSubmit(blknum uint32) {
	go func() {
		self.bftActionC <- &BftAction{
			Type:     SubmitBlock,
			BlockNum: blknum,
		}
	}()
}

func (self *Server) fetchProposal(blkNum uint32, proposer uint32) error {
	msg := self.constructProposalFetchMsg(blkNum, proposer)
	self.msgSendC <- &SendMsgEvent{
		ToPeer: math.MaxUint32,
		Msg:    msg,
	}
	return nil
}

func (self *Server) handleProposalTimeout(evt *TimerEvent) error {
	if self.blockPool.endorsedForBlock(evt.blockNum) {
		return nil
	}
	if !isReady(self.getState()) {
		return nil
	}
	proposals := self.blockPool.getBlockProposals(evt.blockNum)

	log.Infof("server %d proposal timeout, known proposals %d, timeout: %d", self.Index, len(proposals), evt.evtType)

	// if no proposal available, random backoff
	if len(proposals) == 0 {
		log.Infof("no proposal available for block %d, timeout: %d", evt.blockNum, evt.evtType)

		switch evt.evtType {
		case EventProposeBlockTimeout:
			if err := self.timer.StartBackoffTimer(evt.blockNum); err != nil {
				return fmt.Errorf("failed to blk:%d, startbackofftimer err: %s", evt.blockNum, err)
			}
			log.Infof("server %d started backoff timer for blk %d", self.Index, evt.blockNum)
			return nil
		case EventRandomBackoff:
			if self.is2ndProposer(evt.blockNum, self.Index) {
				if err := self.makeProposal(evt.blockNum, true); err != nil {
					return fmt.Errorf("failed to propose empty block: %s", err)
				}
				if err := self.timer.Start2ndProposalTimer(evt.blockNum); err != nil {
					return fmt.Errorf("failed to propose start2ndproposaltimer err:%s", err)
				}
				log.Infof("server %d proposed empty block for blk %d", self.Index, evt.blockNum)
			}
			return nil
		case EventPropose2ndBlockTimeout:
			// 2nd proposal without any proposal, force resync
			self.restartSyncing()
			return nil
		}
	}

	if evt.evtType == EventRandomBackoff {
		// proposal available, no proposing
		return nil
	}

	// find highest rank proposal
	proposal := self.getHighestRankProposal(evt.blockNum, proposals)
	if proposal != nil {
		if self.isProposer(evt.blockNum, proposal.Block.getProposer()) {
			// proposal msg handler will do the endorsement
			return nil
		}

		self.bftActionC <- &BftAction{
			Type:     EndorseBlock,
			BlockNum: evt.blockNum,
			Proposal: proposal,
			forEmpty: false,
		}
	} else {
		log.Errorf("server: %d, blkNum: %d, failed get better proposal, first proposal: %d, %d",
			self.Index, evt.blockNum, proposals[0].Block.getProposer(), proposals[0].Block.getBlockNum())
	}
	return nil
}

// TODO: refactor this
func (self *Server) catchConsensus(blkNum uint32) error {
	if !self.isEndorser(blkNum, self.Index) && !self.isCommitter(blkNum, self.Index) {
		return nil
	}

	proposals := make(map[uint32]*blockProposalMsg)
	pMsgs := self.msgPool.GetProposalMsgs(blkNum)
	for _, msg := range pMsgs {
		p, ok := msg.(*blockProposalMsg)
		if !ok {
			continue
		}
		proposals[p.Block.getProposer()] = p
	}

	chainCfg := self.GetChainConfig()
	eMsgs := self.msgPool.GetEndorsementsMsgs(blkNum)
	var proposal *blockProposalMsg
	endorseDone := false
	endorseEmpty := false
	if len(eMsgs) > int(chainCfg.C) {
		var maxProposer uint32
		emptyCnt := 0
		maxCnt := 0
		proposers := make(map[uint32]int)
		for _, msg := range eMsgs {
			c, ok := msg.(*blockEndorseMsg)
			if !ok {
				continue
			}
			if c.EndorseForEmpty {
				emptyCnt++
			}
			proposers[c.EndorsedProposer] += 1
			if proposers[c.EndorsedProposer] > maxCnt {
				maxProposer = c.EndorsedProposer
				maxCnt = proposers[c.EndorsedProposer]
			}
		}
		proposal = proposals[maxProposer]
		if maxCnt > int(chainCfg.C) {
			endorseDone = true
		}
		if emptyCnt > int(chainCfg.C) {
			endorseDone = true
			endorseEmpty = true
		}
	}
	if proposal != nil && self.isProposer(blkNum, proposal.Block.getProposer()) {
		self.processProposalMsg(proposal)
	}

	if self.isEndorser(blkNum, self.Index) && !endorseDone && proposal != nil {
		return self.endorseBlock(proposal, endorseEmpty)
	}

	if !endorseDone {
		return fmt.Errorf("server %d catch consensus with endorse failed", self.Index)
	}

	if !self.isCommitter(blkNum, self.Index) {
		return nil
	}

	var maxProposer uint32
	maxCnt := 0
	emptyCnt := 0
	proposers := make(map[uint32]int)
	cMsgs := self.msgPool.GetCommitMsgs(blkNum)
	for _, msg := range cMsgs {
		c, ok := msg.(*blockCommitMsg)
		if !ok {
			continue
		}
		if c.CommitForEmpty {
			emptyCnt++
		}
		proposers[c.BlockProposer] += 1
		if proposers[c.BlockProposer] > maxCnt {
			maxProposer = c.BlockProposer
		}
	}

	if p := proposals[maxProposer]; p != nil {
		return self.commitBlock(p, emptyCnt > 0)
	}

	return nil
}

func (self *Server) verifyPrevBlockHash(blkNum uint32, proposal *blockProposalMsg) error {
	prevBlk, prevBlkHash := self.blockPool.getSealedBlock(blkNum - 1)
	if prevBlk == nil {
		// TODO: has no candidate proposals for prevBlock, should restart syncing
		return fmt.Errorf("failed to get prevBlock of current round (%d)", blkNum)
	}
	prevBlkHash2 := proposal.Block.getPrevBlockHash()
	if prevBlkHash != prevBlkHash2 {
		// continue waiting for more proposals
		// FIXME
		return fmt.Errorf("inconsistent prev-block hash %s vs %s (blk %d)",
			prevBlkHash.ToHexString(), prevBlkHash2.ToHexString(), blkNum)
	}

	return nil
}

func (self *Server) hasBlockConsensused() bool {
	blkNum := self.GetCurrentBlockNo()

	C := int(self.GetChainConfig().C)
	cMsgs := self.msgPool.GetCommitMsgs(blkNum)
	emptyCnt := 0
	proposers := make(map[uint32]int)
	for _, msg := range cMsgs {
		c, ok := msg.(*blockCommitMsg)
		if !ok {
			continue
		}
		if c.CommitForEmpty {
			emptyCnt++
		}
		proposers[c.BlockProposer] += 1
		if proposers[c.BlockProposer] > C {
			return true
		}
	}

	return emptyCnt > C
}

func (self *Server) restartSyncing() {

	// send sync request to self.sync, go syncing-state immediately
	// stop all bft timers

	self.stateMgr.StateEventC <- &StateEvent{
		Type:     ForceCheckSync,
		blockNum: self.GetCommittedBlockNo(),
	}
}