viewchange_mgr.go

// Copyright 2016-2017 Hyperchain Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//    http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package rbft

import (
	"context"
	"encoding/hex"
	"sort"
	"time"

	"github.com/samber/lo"
	"golang.org/x/crypto/sha3"

	"github.com/axiomesh/axiom-bft/common"
	"github.com/axiomesh/axiom-bft/common/consensus"
	"github.com/axiomesh/axiom-bft/common/metrics"
	"github.com/axiomesh/axiom-bft/types"
	"github.com/axiomesh/axiom-kit/txpool"
)

var (
	// initialNewView is the first new view cert in every epoch which indicates view = 0.
	// However, the first new view cert is not generated by quorum view changes, so, for
	// every new epoch, we need to initialize a cluster view change immediately after epoch
	// change to advance view to 1 or higher, generate a valid and provable new view.
	initialNewView = &consensus.NewView{}
)

// vcManager manages the whole process of view change
type vcManager struct {
	viewChangePeriod   uint64        //  automatic vc period, default to 0 to disable automatic vc
	viewChangeSeqNo    uint64        // next seqNo to perform view change
	lastNewViewTimeout time.Duration // last timeout we used during this view change
	newViewTimerReason string        // what triggered the timer
	vcHandled          bool          // if we have finished process new view or not

	qlist           map[qidx]*consensus.VcPq        // store Pre-Prepares for view change
	plist           map[uint64]*consensus.VcPq      // store Prepares for view change
	newViewStore    map[uint64]*consensus.NewView   // track last new-view we received or sent
	newViewCache    map[newViewIdx]*newViewCert     // cache all correct new-view we received
	viewChangeStore map[vcIdx]*consensus.ViewChange // track view-change messages

	// track latest QuorumViewChange
	latestQuorumViewChange *quorumViewChangeCache

	// track latest NewView
	latestNewView *consensus.NewView

	// track higher view from other nodes, map node id to view.
	higherViewRecord map[uint64]uint64

	logger common.Logger

	continuousNullRequestCounter uint64
	lastNullRequestSeqNo         uint64
}

// quorumViewChangeCache is the cache of a QuorumViewChange message.
type quorumViewChangeCache struct {
	targetView uint64
	*consensus.QuorumViewChange
}

// newViewIdx indicates a unique NewView message used to help single node recovery, which contains
// 1. target view to help lagging node recover to a correct view
// 2. new view hash to ensure a weak cert NewView
// 3. initial checkpoint height to help lagging node recover to a correct checkpoint height
type newViewIdx struct {
	targetView             uint64
	newViewHash            string
	initialCheckpointState types.CheckpointState
}

// newViewCert is the certification for a new view.
type newViewCert struct {
	forwardPeers       []uint64
	initialCheckpoints []*consensus.SignedCheckpoint
}

// dispatchViewChangeMsg dispatches view change consensus messages from
// other peers And push them into corresponding function
func (rbft *rbftImpl[T, Constraint]) dispatchViewChangeMsg(e consensusEvent) consensusEvent {
	switch et := e.(type) {
	case *consensus.ViewChange:
		return rbft.recvViewChange(et, false)
	case *consensus.NewView:
		return rbft.recvNewView(et)
	case *consensus.FetchBatchRequest:
		return rbft.recvFetchRequestBatch(et)
	case *consensus.FetchBatchResponse:
		return rbft.recvFetchBatchResponse(et)
	case *consensus.FetchView:
		return rbft.recvFetchView(et)
	case *consensus.RecoveryResponse:
		return rbft.recvRecoveryResponse(et)
	case *consensus.QuorumViewChange:
		return rbft.recvQuorumViewChange(et)
	}
	return nil
}

// newVcManager init a instance of view change manager and initialize each parameter
// according to the configuration file.
func newVcManager(c Config) *vcManager {
	vcm := &vcManager{
		qlist:            make(map[qidx]*consensus.VcPq),
		plist:            make(map[uint64]*consensus.VcPq),
		newViewStore:     make(map[uint64]*consensus.NewView),
		newViewCache:     make(map[newViewIdx]*newViewCert),
		viewChangeStore:  make(map[vcIdx]*consensus.ViewChange),
		higherViewRecord: make(map[uint64]uint64),
		logger:           c.Logger,
	}

	vcm.lastNewViewTimeout = c.NewViewTimeout

	return vcm
}

func (rbft *rbftImpl[T, Constraint]) punishAbnormalNodes(isTemp bool, isRecovery bool, view uint64, validatorDynamicInfoMap map[uint64]*ValidatorInfo) {
	if view == 0 || view == rbft.chainConfig.LastStableView || view < rbft.chainConfig.View {
		return
	}
	if !isTemp {
		rbft.logger.Infof("Replica %d try punish abnormal nodes, LastStableView: %d, pre view: %d, now view: %d", rbft.chainConfig.SelfID, rbft.chainConfig.LastStableView, rbft.chainConfig.View, view)
	}

	punishAbnormalNode := func(v uint64) {
		abnormalNodeID := rbft.chainConfig.calPrimaryIDByView(v, validatorDynamicInfoMap)
		if validatorDynamicInfoMap[abnormalNodeID].ConsensusVotingPowerReduced && validatorDynamicInfoMap[abnormalNodeID].ConsensusVotingPowerReduceView == v {
			// already reduce
			return
		}
		if !isTemp {
			rbft.logger.Warningf("Replica %d reduce the voting power of abnormal node %d, view %d", rbft.chainConfig.SelfID, abnormalNodeID, v)
		}
		validatorDynamicInfoMap[abnormalNodeID].ConsensusVotingPowerReduced = true
		validatorDynamicInfoMap[abnormalNodeID].ConsensusVotingPowerReduceView = v
		validatorDynamicInfoMap[abnormalNodeID].ConsensusVotingPower = rbft.chainConfig.EpochInfo.ConsensusParams.NotActiveWeight
	}
	// There may be multiple rounds of viewchange, so it is necessary to reduce the selected primary nodes in each round.
	for v := rbft.chainConfig.LastStableView; v < view; v++ {
		// if recover doesn't reduce the last view node
		if v == view-1 && isRecovery {
			continue
		}
		punishAbnormalNode(v)
	}
}

func (rbft *rbftImpl[T, Constraint]) recoverAbnormalNodes(view uint64, validatorDynamicInfoMap map[uint64]*ValidatorInfo) {
	if view == 0 || view < rbft.chainConfig.View {
		return
	}

	// recover abnormal node voting power
	for id, n := range validatorDynamicInfoMap {
		if n.ConsensusVotingPowerReduced && view >= n.ConsensusVotingPowerReduceView+rbft.chainConfig.EpochInfo.ConsensusParams.AbnormalNodeExcludeView {
			rbft.logger.Infof("Replica %d recover abnormal node %d voting power from %d to %d, reduce view: %d", rbft.chainConfig.SelfID, id, n.ConsensusVotingPower, rbft.chainConfig.ValidatorSet[id], n.ConsensusVotingPowerReduceView)
			n.ConsensusVotingPower = rbft.chainConfig.ValidatorSet[id]
			n.ConsensusVotingPowerReduced = false
			n.ConsensusVotingPowerReduceView = 0
		}
	}
}

// setView sets the view with the viewLock.
func (rbft *rbftImpl[T, Constraint]) setView(view uint64) {
	rbft.viewLock.Lock()
	if !rbft.isTest {
		unstableValidatorDynamicInfoMap := lo.MapEntries(rbft.chainConfig.LastStableValidatorDynamicInfoMap, func(k uint64, v *ValidatorInfo) (uint64, *ValidatorInfo) {
			return k, &ValidatorInfo{
				ID:                             v.ID,
				ConsensusVotingPower:           v.ConsensusVotingPower,
				ConsensusVotingPowerReduced:    v.ConsensusVotingPowerReduced,
				ConsensusVotingPowerReduceView: v.ConsensusVotingPowerReduceView,
			}
		})
		rbft.punishAbnormalNodes(false, false, view, unstableValidatorDynamicInfoMap)
		rbft.recoverAbnormalNodes(view, unstableValidatorDynamicInfoMap)
		rbft.chainConfig.ValidatorDynamicInfoMap = unstableValidatorDynamicInfoMap
	}
	rbft.chainConfig.View = view
	rbft.chainConfig.updatePrimaryID()
	rbft.viewLock.Unlock()
	rbft.metrics.viewGauge.Set(float64(view))
}

func (rbft *rbftImpl[T, Constraint]) setViewWithRecovery(view uint64) {
	rbft.viewLock.Lock()
	if !rbft.isTest {
		unstableValidatorDynamicInfoMap := lo.MapEntries(rbft.chainConfig.LastStableValidatorDynamicInfoMap, func(k uint64, v *ValidatorInfo) (uint64, *ValidatorInfo) {
			return k, &ValidatorInfo{
				ID:                             v.ID,
				ConsensusVotingPower:           v.ConsensusVotingPower,
				ConsensusVotingPowerReduced:    v.ConsensusVotingPowerReduced,
				ConsensusVotingPowerReduceView: v.ConsensusVotingPowerReduceView,
			}
		})
		rbft.punishAbnormalNodes(false, true, view, unstableValidatorDynamicInfoMap)
		rbft.recoverAbnormalNodes(view, unstableValidatorDynamicInfoMap)
		rbft.chainConfig.ValidatorDynamicInfoMap = unstableValidatorDynamicInfoMap
	}
	rbft.chainConfig.View = view
	rbft.chainConfig.updatePrimaryID()
	rbft.viewLock.Unlock()
	rbft.metrics.viewGauge.Set(float64(view))
}

// getUnstableValidatorDynamicInfoMap returns the unstable validator dynamic info
// assumes that the abnormal node is restored after the view is stable
func (rbft *rbftImpl[T, Constraint]) getUnstableValidatorDynamicInfoMap(view uint64, isRecovery bool) []*consensus.NodeDynamicInfo {
	unstableValidatorDynamicInfoMap := lo.MapEntries(rbft.chainConfig.LastStableValidatorDynamicInfoMap, func(k uint64, v *ValidatorInfo) (uint64, *ValidatorInfo) {
		return k, &ValidatorInfo{
			ID:                             v.ID,
			ConsensusVotingPower:           v.ConsensusVotingPower,
			ConsensusVotingPowerReduced:    v.ConsensusVotingPowerReduced,
			ConsensusVotingPowerReduceView: v.ConsensusVotingPowerReduceView,
		}
	})
	info := lo.MapToSlice(unstableValidatorDynamicInfoMap, func(id uint64, v *ValidatorInfo) ValidatorInfo {
		return ValidatorInfo{
			ID:                             v.ID,
			ConsensusVotingPower:           v.ConsensusVotingPower,
			ConsensusVotingPowerReduced:    v.ConsensusVotingPowerReduced,
			ConsensusVotingPowerReduceView: v.ConsensusVotingPowerReduceView,
		}
	})
	sort.Slice(info, func(i, j int) bool {
		return info[i].ID < info[j].ID
	})
	rbft.punishAbnormalNodes(true, isRecovery, view, unstableValidatorDynamicInfoMap)
	rbft.recoverAbnormalNodes(view, unstableValidatorDynamicInfoMap)
	res := lo.MapToSlice(unstableValidatorDynamicInfoMap, func(id uint64, item *ValidatorInfo) *consensus.NodeDynamicInfo {
		return &consensus.NodeDynamicInfo{
			Id:                             item.ID,
			ConsensusVotingPower:           item.ConsensusVotingPower,
			ConsensusVotingPowerReduced:    item.ConsensusVotingPowerReduced,
			ConsensusVotingPowerReduceView: item.ConsensusVotingPowerReduceView,
		}
	})
	sort.Slice(res, func(i, j int) bool {
		return res[i].Id < res[j].Id
	})
	return res
}

// setLastStableView after checkpoint or accept newView
func (rbft *rbftImpl[T, Constraint]) setLastStableView(view uint64) {
	rbft.chainConfig.LastStableView = view
}

func (rbft *rbftImpl[T, Constraint]) checkView(msg *consensus.ConsensusMessage) {
	if !rbft.chainConfig.CheckValidator(msg.From) {
		return
	}
	// record higher view to actively fetch view periodically.
	if msg.View > rbft.chainConfig.View {
		rbft.vcMgr.higherViewRecord[msg.From] = msg.View
	}
}

func (rbft *rbftImpl[T, Constraint]) initRecovery() consensusEvent {
	return rbft.sendViewChange(true)
}

// sendViewChange broadcasts view change message to all other replicas to request
// all other replicas' current status, there will be two cases which need to trigger
// view change:
//  1. general view change: current node finds current primary behavior incorrectly.
//  2. recovery view change: current node finds itself primary behavior incorrectly, such as
//     2.1 restart node and trigger view change
//     2.2 epoch changed and trigger view change
//     2.3 view change resend timer out and trigger view change again
//     2.4 find self inconsistent with quorum nodes in sync state.
func (rbft *rbftImpl[T, Constraint]) sendViewChange(status ...bool) consensusEvent {
	recovery := false
	if len(status) != 0 {
		recovery = status[0]
	}

	// reject view change when current node is in StateTransferring to avoid other (quorum-1) replicas
	// finish view change but cannot reach consensus later if there are only quorum active replicas
	// consisting of (quorum-1) normal replicas + 1 lagging replica.
	if rbft.atomicIn(StateTransferring) {
		rbft.logger.Warningf("Replica %d is in state transferring, not sending view change", rbft.chainConfig.SelfID)
		return nil
	}

	// do some check and do some preparation
	// such as stop nullRequest timer, clean vcMgr.viewChangeStore and so on.
	err := rbft.beforeSendVC(recovery)
	if err != nil {
		return nil
	}

	if recovery {
		rbft.atomicOn(InRecovery)
		rbft.metrics.statusGaugeInRecovery.Set(InRecovery)
	}

	// create viewChange message
	vcBasis := rbft.getVcBasis()
	vc := &consensus.ViewChange{
		Basis:    vcBasis,
		Recovery: recovery,
	}
	sig, sErr := rbft.signViewChange(vc)
	if sErr != nil {
		rbft.logger.Warningf("Replica %d sign view change failed: %s", rbft.chainConfig.SelfID, sErr)
		return nil
	}
	vc.Signature = sig

	rbft.logger.Infof("Replica %d sending viewChange, v:%d, h:%d, |C|:%d, |P|:%d, "+
		"|Q|:%d, recovery: %+v", rbft.chainConfig.SelfID, vcBasis.GetView(), vcBasis.GetH(), len(vcBasis.GetCset()),
		len(vcBasis.GetPset()), len(vcBasis.GetQset()), recovery)
	rbft.logger.Debugf("Replica %d sending viewChange, if recover ValidatorDynamicInfo: %v", rbft.chainConfig.SelfID, showSimpleValidatorDynamicInfo(vc.Basis.IfRecoverValidatorDynamicInfo))
	rbft.logger.Debugf("Replica %d sending viewChange, if not recover ValidatorDynamicInfo: %v", rbft.chainConfig.SelfID, showSimpleValidatorDynamicInfo(vc.Basis.IfNotRecoverValidatorDynamicInfo))
	payload, err := vc.MarshalVTStrict()
	if err != nil {
		rbft.logger.Errorf("ConsensusMessage_VIEW_CHANGE Marshal Error: %s", err)
		return nil
	}
	consensusMsg := &consensus.ConsensusMessage{
		Type:    consensus.Type_VIEW_CHANGE,
		Payload: payload,
	}
	rbft.peerMgr.broadcast(context.TODO(), consensusMsg)

	rbft.logger.Trace(consensus.TagNameViewChange, consensus.TagStageStart, consensus.TagContentViewChange{
		Node: rbft.chainConfig.SelfID,
		View: rbft.chainConfig.View,
	})

	event := &LocalEvent{
		Service:   ViewChangeService,
		EventType: ViewChangeResendTimerEvent,
	}
	// start vcResendTimer. If peers can't viewChange successfully within the given time,
	// enter recovery view change.
	rbft.timerMgr.startTimer(vcResendTimer, event)
	// start fetch view timer to actively fetch view periodically.
	rbft.startFetchViewTimer()
	return rbft.recvViewChange(vc, true)
}

// recvViewChange processes ViewChange message from itself or other peers
// if the number of ViewChange message for the same view reach on commonCaseQuorum, return ViewChangeQuorumEvent.
// else, peers may resend vc or wait more vc message arrived.
func (rbft *rbftImpl[T, Constraint]) recvViewChange(vc *consensus.ViewChange, verified bool) consensusEvent {
	vcBasis := vc.Basis
	remoteReplicaID := vcBasis.GetReplicaId()
	targetView := vcBasis.GetView()

	rbft.logger.Infof("Replica %d received viewChange from replica %d, v:%d, h:%d, "+
		"|C|:%d, |P|:%d, |Q|:%d, recovery: %v", rbft.chainConfig.SelfID, remoteReplicaID, targetView, vcBasis.GetH(),
		len(vcBasis.GetCset()), len(vcBasis.GetPset()), len(vcBasis.GetQset()), vc.Recovery)
	rbft.logger.Debugf("Replica %d received viewChange from replica %d, if recover ValidatorDynamicInfo: %v", rbft.chainConfig.SelfID, remoteReplicaID, showSimpleValidatorDynamicInfo(vc.Basis.IfRecoverValidatorDynamicInfo))
	rbft.logger.Debugf("Replica %d received viewChange from replica %d, if not recover ValidatorDynamicInfo: %v", rbft.chainConfig.SelfID, remoteReplicaID, showSimpleValidatorDynamicInfo(vc.Basis.IfNotRecoverValidatorDynamicInfo))

	if targetView < rbft.chainConfig.View {
		// for recovery node with a lower view, we can directly send back latest new view to help
		// remote node advance to the latest new view if we are in normal status.
		if vc.Recovery && rbft.isNormal() {
			rbft.logger.Warningf("Replica %d found viewChange message for old view from "+
				"replica %d, help remote recovery", rbft.chainConfig.SelfID, remoteReplicaID)
			return rbft.sendRecoveryResponse(remoteReplicaID, targetView)
		}

		rbft.logger.Warningf("Replica %d found viewChange message for old view from "+
			"replica %d: self view=%d, vc view=%d", rbft.chainConfig.SelfID, remoteReplicaID, rbft.chainConfig.View, targetView)
		return nil
	}

	// check view change correctness
	if !rbft.correctViewChange(vcBasis) {
		rbft.logger.Warningf("Replica %d found viewChange message incorrect", rbft.chainConfig.SelfID)
		return nil
	}

	// check if this view change has stored in viewChangeStore, if so, return nil
	if old, ok := rbft.vcMgr.viewChangeStore[vcIdx{v: targetView, id: remoteReplicaID}]; ok {
		if !vc.Recovery && old.Basis.EqualVT(vc.Basis) {
			rbft.logger.Warningf("Replica %d already has a same viewChange message"+
				" for view %d from replica %d, ignore it", rbft.chainConfig.SelfID, targetView, remoteReplicaID)
			return nil
		}

		rbft.logger.Debugf("Replica %d already has an updated viewChange message"+
			" for view %d from replica %d, replace it", rbft.chainConfig.SelfID, targetView, remoteReplicaID)
	}

	// verify view change signature if not verified.
	if !verified {
		vErr := rbft.verifySignedViewChange(vc, remoteReplicaID)
		if vErr != nil {
			rbft.logger.Errorf("Replica %d found invalid view change message, error: %s", rbft.chainConfig.SelfID, vErr)
			return nil
		}
	}

	vc.Timestamp = time.Now().UnixNano()
	if rbft.chainConfig.CheckValidator(remoteReplicaID) {
		// store vc to viewChangeStore
		rbft.vcMgr.viewChangeStore[vcIdx{v: targetView, id: remoteReplicaID}] = vc
	} else {
		rbft.logger.Debugf("Replica %d not accept viewchane msg from non-validator %d",
			rbft.chainConfig.SelfID, remoteReplicaID)
	}

	rbft.logger.Trace(consensus.TagNameViewChange, consensus.TagStageReceive, consensus.TagContentViewChange{
		Node: remoteReplicaID,
		View: targetView,
	})

	// RBFT TOCS 4.5.1 Liveness: "if a replica receives a set of
	// f+1 valid VIEW-CHANGE messages from other replicas for
	// views greater than its current view, it sends a VIEW-CHANGE
	// message for the smallest view in the set, even if its timer
	// has not expired"
	replicas := make(map[uint64]bool)
	minView := uint64(0)
	for idx, remoteVC := range rbft.vcMgr.viewChangeStore {
		if remoteVC.Timestamp+int64(rbft.timerMgr.getTimeoutValue(cleanViewChangeTimer)) < time.Now().UnixNano() {
			rbft.logger.Debugf("Replica %d drop an out-of-time viewChange message from replica %d",
				rbft.chainConfig.SelfID, idx.id)
			delete(rbft.vcMgr.viewChangeStore, idx)
			continue
		}

		if idx.v <= rbft.chainConfig.View {
			continue
		}

		replicas[idx.id] = true
		if minView == 0 || idx.v < minView {
			minView = idx.v
		}
	}

	// we only enter this if there are enough view change messages greater than our current view
	if len(replicas) >= rbft.oneCorrectQuorum() {
		rbft.logger.Infof("Replica %d received f+1 viewChange messages whose view is greater than "+
			"current view %d, detailed: %v, triggering viewChange to view %d", rbft.chainConfig.SelfID, rbft.chainConfig.View, replicas, minView)
		// subtract one, because sendViewChange() increments
		newView := minView - uint64(1)
		rbft.setView(newView)
		return rbft.sendViewChange()
	}
	// calculate how many peers has view = rbft.chainConfig.View
	quorum := 0
	qvc := make([]*consensus.ViewChange, 0, len(rbft.vcMgr.viewChangeStore))
	for idx, vcs := range rbft.vcMgr.viewChangeStore {
		if idx.v == rbft.chainConfig.View {
			quorum++
			qvc = append(qvc, vcs)
		}
	}
	rbft.logger.Debugf("Replica %d now has %d viewChange requests for view %d",
		rbft.chainConfig.SelfID, quorum, rbft.chainConfig.View)

	// if in viewChange and vc.view = rbft.chainConfig.View and quorum >= commonCaseQuorum,
	// jump into ViewChangeQuorumEvent
	if rbft.atomicIn(InViewChange) && targetView == rbft.chainConfig.View && quorum >= rbft.commonCaseQuorum() {
		// close vcResendTimer
		rbft.timerMgr.stopTimer(vcResendTimer)

		// start newViewTimer and increase lastNewViewTimeout.
		// if this view change failed, next viewChange will have more time to do it
		// !!!NOTICE: only reset newViewTimer for the first time we reach the QuorumViewChange
		rbft.softStartNewViewTimer(rbft.vcMgr.lastNewViewTimeout, "new viewChange", true)
		rbft.vcMgr.lastNewViewTimeout = 2 * rbft.vcMgr.lastNewViewTimeout
		if rbft.vcMgr.lastNewViewTimeout > 5*rbft.timerMgr.getTimeoutValue(newViewTimer) {
			rbft.vcMgr.lastNewViewTimeout = 5 * rbft.timerMgr.getTimeoutValue(newViewTimer)
		}

		rbft.vcMgr.latestQuorumViewChange = &quorumViewChangeCache{
			targetView: targetView,
			QuorumViewChange: &consensus.QuorumViewChange{
				ReplicaId:   rbft.chainConfig.SelfID,
				ViewChanges: qvc,
			},
		}
		// packaging ViewChangeQuorumEvent message
		return &LocalEvent{
			Service:   ViewChangeService,
			EventType: ViewChangeQuorumEvent,
		}
	}

	if rbft.isNormal() {
		if rbft.chainConfig.isProposerElectionTypeWRF() {
			if vc.Recovery {
				rbft.logger.Warningf("Replica %d found viewChange message for higher view from "+
					"replica %d, help remote recovery", rbft.chainConfig.SelfID, remoteReplicaID)
				// for recovery node with a higher view, we can directly send back latest new view to help
				// remote node rollback to the latest new view if we are in normal status.
				return rbft.sendRecoveryResponse(remoteReplicaID, targetView)
			}

			// if received vc message from primary, send view change to other peers directly
			if rbft.isPrimary(remoteReplicaID) {
				rbft.logger.Infof("Replica %d received viewChange from old primary %d for view %d, "+
					"trigger viewChange.", rbft.chainConfig.SelfID, remoteReplicaID, targetView)
				return rbft.sendViewChange()
			}
		} else {
			// if received vc message from primary, send view change to other peers directly
			if rbft.isPrimary(remoteReplicaID) {
				rbft.logger.Infof("Replica %d received viewChange from old primary %d for view %d, "+
					"trigger viewChange.", rbft.chainConfig.SelfID, remoteReplicaID, targetView)
				return rbft.sendViewChange()
			}

			if vc.Recovery {
				rbft.logger.Warningf("Replica %d found viewChange message for higher view from "+
					"replica %d, help remote recovery", rbft.chainConfig.SelfID, remoteReplicaID)
				// for recovery node with a higher view, we can directly send back latest new view to help
				// remote node rollback to the latest new view if we are in normal status.
				return rbft.sendRecoveryResponse(remoteReplicaID, targetView)
			}
		}
	}

	return nil
}

func (rbft *rbftImpl[T, Constraint]) recvQuorumViewChange(qvc *consensus.QuorumViewChange) consensusEvent {
	if len(qvc.ViewChanges) == 0 {
		return nil
	}
	// TODO(DH): reject or not?
	if rbft.atomicIn(StateTransferring) {
		rbft.logger.Debugf("Replica %d reject QuorumViewChange as we are in state transfer", rbft.chainConfig.SelfID)
		return nil
	}

	var (
		remoteReplicaID = qvc.GetReplicaId()
		targetView      uint64
	)
	// ensure all ViewChanges have the same target view and target view is not lower than
	// current view.
	for i, vc := range qvc.ViewChanges {
		vcBasis := vc.Basis
		view := vcBasis.GetView()
		if i == 0 {
			targetView = view
			if targetView < rbft.chainConfig.View {
				rbft.logger.Warningf("Replica %d received an invalid QuorumViewChange from %d "+
					"with a target view %d lower than current view %d", rbft.chainConfig.SelfID, remoteReplicaID,
					targetView, rbft.chainConfig.View)
				return nil
			}
			continue
		}
		if view != targetView {
			rbft.logger.Warningf("Replica %d received an invalid QuorumViewChange from %d with "+
				"mismatch target view %d and %d", rbft.chainConfig.SelfID, remoteReplicaID, view, targetView)
			return nil
		}
	}

	rbft.logger.Debugf("Replica %d received quorum view change from %d with target view %d, current view %d",
		rbft.chainConfig.SelfID, remoteReplicaID, targetView, rbft.chainConfig.View)

	// only process ViewChanges with higher view when node is in normal.
	if rbft.isNormal() {
		if targetView > rbft.chainConfig.View {
			rbft.logger.Debugf("Replica %d process quorum view change in normal", rbft.chainConfig.SelfID)
			rbft.processQuorumViewChange(qvc.ViewChanges, false)
		}
		return nil
	}

	// process ViewChanges with higher/equal view when node is in ViewChange status.
	if rbft.atomicIn(InViewChange) {
		if targetView >= rbft.chainConfig.View {
			rbft.logger.Debugf("Replica %d process quorum view change in view change", rbft.chainConfig.SelfID)
			rbft.processQuorumViewChange(qvc.ViewChanges, false)
		}
		return nil
	}

	return nil
}

// processQuorumViewChange loop process all view changes.
func (rbft *rbftImpl[T, Constraint]) processQuorumViewChange(vcs []*consensus.ViewChange, verified bool) {
	for _, vc := range vcs {
		// loop process potential events caused by receive view change.
		if vc.Basis.GetReplicaId() == rbft.chainConfig.SelfID {
			rbft.logger.Debugf("Replica %d reject process view change from itself", rbft.chainConfig.SelfID)
			continue
		}
		next := rbft.recvViewChange(vc, verified)
		if next != nil {
			for {
				next = rbft.processEvent(next)
				if next == nil {
					break
				}
			}
		}
	}
}

// sendNewView select suitable pqc from viewChangeStore as a new view message and
// broadcast to replica peers when peer is primary
func (rbft *rbftImpl[T, Constraint]) sendNewView() consensusEvent {
	// if this new view has stored, return nil.
	if _, ok := rbft.vcMgr.newViewStore[rbft.chainConfig.View]; ok {
		rbft.logger.Warningf("Replica %d already has newView in store for view %d, ignore it",
			rbft.chainConfig.SelfID, rbft.chainConfig.View)
		return nil
	}
	var (
		vcSet *consensus.QuorumViewChange
		basis []*consensus.VcBasis
	)
	vcSet, basis = rbft.getViewChangeBasis()

	validatorDynamicInfo, ok := rbft.selectValidatorDynamicInfo(vcSet.ViewChanges)
	if !ok {
		rbft.logger.Infof("Replica %d could not find consistent validatorDynamicInfo",
			rbft.chainConfig.SelfID)
		for _, vc := range vcSet.ViewChanges {
			rbft.logger.Infof("Replica %d view change IfRecoverValidatorDynamicInfo: %v, IfNotRecoverValidatorDynamicInfo: %v", vc.Basis.ReplicaId, showSimpleValidatorDynamicInfo(vc.Basis.IfRecoverValidatorDynamicInfo), showSimpleValidatorDynamicInfo(vc.Basis.IfNotRecoverValidatorDynamicInfo))
		}
		return nil
	}

	// get suitable checkpoint for later recovery, replicas contains the peer no who has this checkpoint.
	// if we can't find suitable checkpoint, ok return false.
	checkpointState, checkpointSet, ok := rbft.selectInitialCheckpoint(basis)
	if !ok {
		rbft.logger.Infof("Replica %d could not find consistent checkpoint: %+v",
			rbft.chainConfig.SelfID, rbft.vcMgr.viewChangeStore)
		return nil
	}
	rbft.logger.Debugf("initial checkpoint: %+v", checkpointState)
	// select suitable pqcCerts for later recovery. Their sequence is greater than cp
	// if msgList is nil, must some bug happened
	msgList := rbft.assignSequenceNumbers(basis, checkpointState.Meta.Height)
	if msgList == nil {
		rbft.logger.Infof("Replica %d could not assign sequence numbers for newView", rbft.chainConfig.SelfID)
		return nil
	}
	rbft.logger.Debugf("x-set: %+v", msgList)
	// create new view message
	nv := &consensus.NewView{
		View:          rbft.chainConfig.View,
		Xset:          msgList,
		ReplicaId:     rbft.chainConfig.SelfID,
		ViewChangeSet: vcSet,
		QuorumCheckpoint: &consensus.QuorumCheckpoint{
			Checkpoint: &consensus.Checkpoint{
				ExecuteState: &consensus.Checkpoint_ExecuteState{
					Height: checkpointState.Meta.Height,
					Digest: checkpointState.Meta.Digest,
				},
			},
		},
		FromId:               rbft.chainConfig.SelfID,
		ValidatorDynamicInfo: validatorDynamicInfo,
	}
	sig, sErr := rbft.signNewView(nv)
	if sErr != nil {
		rbft.logger.Warningf("Replica %d sign new view failed: %s", rbft.chainConfig.SelfID, sErr)
		return nil
	}
	nv.Signature = sig

	// check if primary need state update
	needStateUpdate := rbft.checkIfNeedStateUpdate(checkpointState, checkpointSet)
	if needStateUpdate || rbft.atomicIn(StateTransferring) {
		rbft.logger.Debugf("Primary %d needs to catch up in viewChange", rbft.chainConfig.SelfID)
		return nil
	}

	rbft.logger.Infof("Replica %d is new primary, sending newView, v:%d, X:%+v",
		rbft.chainConfig.SelfID, nv.View, msgList)
	rbft.logger.Infof("Replica %d generated newView validatorDynamicInfo: %v",
		rbft.chainConfig.SelfID, showSimpleValidatorDynamicInfo(validatorDynamicInfo))
	payload, err := nv.MarshalVTStrict()
	if err != nil {
		rbft.logger.Errorf("ConsensusMessage_NEW_VIEW Marshal Error: %s", err)
		return nil
	}
	consensusMsg := &consensus.ConsensusMessage{
		Type:    consensus.Type_NEW_VIEW,
		Payload: payload,
	}
	// broadcast new view
	rbft.peerMgr.broadcast(context.TODO(), consensusMsg)
	// set new view to newViewStore
	rbft.vcMgr.newViewStore[rbft.chainConfig.View] = nv

	return rbft.primaryCheckNewView(msgList)
}

// recvNewView receives new view message and check if this node could
// process this message or not.
func (rbft *rbftImpl[T, Constraint]) recvNewView(nv *consensus.NewView) consensusEvent {
	targetView, _, valid := rbft.checkNewView(nv)
	if valid {
		// direct process ViewChanges and then process new view.
		rbft.processQuorumViewChange(nv.ViewChangeSet.ViewChanges, true)

		// latestNewView mey be changed during above process(process vc -> primary generate new view ->
		// update latestNewView), so we need to ensure not process the same new view again.
		if rbft.vcMgr.latestNewView.View >= targetView {
			rbft.logger.Noticef("Replica %d has advanced to view %d, ignore process new view %d",
				rbft.chainConfig.SelfID, rbft.vcMgr.latestNewView.View, targetView)
			return nil
		}

		rbft.vcMgr.newViewStore[targetView] = nv
		return rbft.replicaCheckNewView()
	}
	return nil
}

// checkNewView checks the validity of new view message and returns new view info:
// 1. new view target view
// 2. new view hash
// 3. vc basis of new view
// 4. if new view is valid or not
func (rbft *rbftImpl[T, Constraint]) checkNewView(nv *consensus.NewView) (uint64, string, bool) {
	rbft.logger.Infof("Replica %d received newView %d from replica %d", rbft.chainConfig.SelfID,
		nv.View, nv.ReplicaId)

	// view 0 is the initial view of every epoch, but valid NewView should start from 1.
	if nv.View == 0 {
		rbft.logger.Debugf("Replica %d reject invalid newView 0", rbft.chainConfig.SelfID)
		return 0, "", false
	}

	// TODO(DH): reject or not?
	if rbft.atomicIn(StateTransferring) {
		rbft.logger.Debugf("Replica %d reject newView as we are in state transfer", rbft.chainConfig.SelfID)
		return 0, "", false
	}

	// wrf recovery cannot check the PrimaryID
	if !rbft.chainConfig.isProposerElectionTypeWRF() {
		expectedPrimaryID := rbft.chainConfig.calPrimaryIDByView(nv.View, rbft.chainConfig.ValidatorDynamicInfoMap)
		if expectedPrimaryID != nv.ReplicaId {
			rbft.logger.Warningf("Replica %d reject invalid newView from %d, v:%d, expected primary: %d",
				rbft.chainConfig.SelfID, nv.ReplicaId, nv.View, expectedPrimaryID)

			rbft.logger.Debugf("Replica %d current height: %d, dig: %s",
				rbft.chainConfig.SelfID, rbft.exec.lastExec, rbft.chainConfig.LastCheckpointExecBlockHash)
			return 0, "", false
		}
	}

	var (
		targetView uint64
	)

	// ensure all ViewChanges have the same target view and target view is equal to
	// NewView.view.
	for i, vc := range nv.ViewChangeSet.ViewChanges {
		vcBasis := vc.Basis
		view := vcBasis.GetView()
		if i == 0 {
			targetView = view
			if targetView != nv.View {
				rbft.logger.Warningf("Replica %d received an invalid NewView with mismatch target "+
					"view %d and NewView %d", rbft.chainConfig.SelfID, targetView, nv.View)
				return 0, "", false
			}
			continue
		}
		if view != targetView {
			rbft.logger.Warningf("Replica %d received an invalid NewView with mismatch target "+
				"view %d and %d", rbft.chainConfig.SelfID, view, targetView)
			return 0, "", false
		}
	}
	validatorDynamicInfo, ok := rbft.selectValidatorDynamicInfo(nv.ViewChangeSet.ViewChanges)
	if !ok {
		rbft.logger.Infof("Replica %d received an invalid NewView that could not find consistent validatorDynamicInfo",
			rbft.chainConfig.SelfID)
		for _, vc := range nv.ViewChangeSet.ViewChanges {
			rbft.logger.Infof("Replica %d view change IfRecoverValidatorDynamicInfo: %v, IfNotRecoverValidatorDynamicInfo: %v", vc.Basis.ReplicaId, showSimpleValidatorDynamicInfo(vc.Basis.IfRecoverValidatorDynamicInfo), showSimpleValidatorDynamicInfo(vc.Basis.IfNotRecoverValidatorDynamicInfo))
		}
		return 0, "", false
	}

	if !consensus.ValidatorDynamicInfoEqual(validatorDynamicInfo, nv.ValidatorDynamicInfo) {
		rbft.logger.Warningf("Replica %d received an invalid NewView with mismatch validatorDynamicInfo, viewchange's: %v, newView's: %v", rbft.chainConfig.SelfID, showSimpleValidatorDynamicInfo(validatorDynamicInfo), showSimpleValidatorDynamicInfo(nv.ValidatorDynamicInfo))
		return 0, "", false
	}

	// verify signature of new view.
	nvHash, vErr := rbft.verifySignedNewView(nv)
	if vErr != nil {
		rbft.logger.Errorf("Replica %d found invalid new view message, error: %s", rbft.chainConfig.SelfID, vErr)
		return 0, "", false
	}

	// verify the signatures of all view changes contained in new view are all correct.
	for _, vc := range nv.ViewChangeSet.ViewChanges {
		vErr = rbft.verifySignedViewChange(vc, vc.Basis.GetReplicaId())
		if vErr != nil {
			rbft.logger.Errorf("Replica %d found invalid view change message in new view, error: %s",
				rbft.chainConfig.SelfID, vErr)
			return 0, "", false
		}
	}

	// if current node is in normal, check the target view
	// 1. targetView > current view, process all view changes and new view to advance view
	// 2. targetView <= current view, may be sent from some out-of-date nodes, ignore it as
	// we are normal in a more advanced view.
	if rbft.isNormal() {
		if targetView > rbft.chainConfig.View {
			return targetView, "", true
		}
		rbft.logger.Warningf("Replica %d reject process new view %d as we "+
			"are in normal", rbft.chainConfig.SelfID, targetView)
		return 0, "", false
	} else if rbft.atomicIn(InRecovery) {
		// if current node is in recovery, direct process new view.
		return targetView, hex.EncodeToString(nvHash), true
	} else if rbft.atomicIn(InViewChange) {
		// if current node is in view change, check the target view
		// 1. targetView >= current view, process all view changes and new view to advance view
		// 2. targetView < current view, which indicates an invalid NewView of current view change
		// progress, ignore it.
		if targetView >= rbft.chainConfig.View {
			return targetView, "", true
		}
		return 0, "", false
	}

	return targetView, "", true
}

// tryFetchView tries fetch view in view change status, compares local view and remote view:
// 1. remoteView > currentView, fetch view
// 2. remoteView <= currentView, ignore
func (rbft *rbftImpl[T, Constraint]) tryFetchView() {
	if !rbft.atomicIn(InViewChange) {
		rbft.logger.Debugf("Replica %d not try to fetch view in normal status", rbft.chainConfig.SelfID)
		return
	}

	// fetch view when remote node has larger view and current node is in view change,
	// ask remote node response new view info to help self advance view.
	for remoteID, remoteView := range rbft.vcMgr.higherViewRecord {
		if remoteView > rbft.chainConfig.View {
			// create fetchView message
			fv := &consensus.FetchView{
				ReplicaId: rbft.chainConfig.SelfID,
				View:      rbft.chainConfig.View,
			}

			rbft.logger.Infof("Replica %d sending fetch view to %d, current view: %d, "+
				"remote view: %d", rbft.chainConfig.SelfID, remoteID, rbft.chainConfig.View, remoteView)

			payload, err := fv.MarshalVTStrict()
			if err != nil {
				rbft.logger.Errorf("ConsensusMessage_FETCH_VIEW Marshal Error: %s", err)
				return
			}
			consensusMsg := &consensus.ConsensusMessage{
				Type:    consensus.Type_FETCH_VIEW,
				Payload: payload,
			}
			rbft.peerMgr.unicast(context.TODO(), consensusMsg, remoteID)
			return
		}
		delete(rbft.vcMgr.higherViewRecord, remoteID)
	}

	rbft.startFetchViewTimer()
}

// recvFetchView helps remote node recover view and height using latestQuorumViewChange and latestNewView cache.
func (rbft *rbftImpl[T, Constraint]) recvFetchView(fv *consensus.FetchView) consensusEvent {
	rbft.logger.Debugf("Replica %d received fetch view from %d, current view: %d, "+
		"remote view: %d", rbft.chainConfig.SelfID, fv.ReplicaId, rbft.chainConfig.View, fv.View)

	if !(rbft.chainConfig.View > fv.View) {
		rbft.logger.Debugf("Replica %d don't need to response fetch view as we don't "+
			"have a larger view", rbft.chainConfig.SelfID)
		return nil
	}

	// 1. check if we have latest QuorumViewChange record, which means we are in ViewChange status
	// and received quorum correct ViewChanges, we can forward this QuorumViewChange to help remote
	// node advance its view.
	if qvc := rbft.vcMgr.latestQuorumViewChange; qvc != nil {
		rbft.logger.Infof("Replica %d sending quorum view change to %d, current view: %d, "+
			"remote view: %d", rbft.chainConfig.SelfID, fv.ReplicaId, rbft.chainConfig.View, fv.View)

		payload, err := qvc.MarshalVTStrict()
		if err != nil {
			rbft.logger.Errorf("ConsensusMessage_QUORUM_VIEW_CHANGE Marshal Error: %s", err)
			return nil
		}
		consensusMsg := &consensus.ConsensusMessage{
			Type:    consensus.Type_QUORUM_VIEW_CHANGE,
			Payload: payload,
		}
		rbft.peerMgr.unicast(context.TODO(), consensusMsg, fv.ReplicaId)
		return nil
	}

	// 2. send back NewView as response to help remote node advance its view if we have no QuorumViewChange
	// cache.
	nv := rbft.vcMgr.latestNewView
	rbft.logger.Infof("Replica %d sending back newView to %d, v:%d", rbft.chainConfig.SelfID, fv.ReplicaId, nv.View)
	payload, err := nv.MarshalVTStrict()
	if err != nil {
		rbft.logger.Errorf("ConsensusMessage_NEW_VIEW Marshal Error: %s", err)
		return nil
	}
	consensusMsg := &consensus.ConsensusMessage{
		Type:    consensus.Type_NEW_VIEW,
		Payload: payload,
	}
	rbft.peerMgr.unicast(context.TODO(), consensusMsg, fv.ReplicaId)

	return nil
}

// sendRecoveryResponse sends recovery response to help remote node recover to a correct state including:
// 1. latest new view info to help remote node advance to a correct view
// 2. latest stable checkpoint info to help remote node advance to a correct initial checkpoint height
func (rbft *rbftImpl[T, Constraint]) sendRecoveryResponse(remoteReplicaID, remoteView uint64) error {
	nv := rbft.vcMgr.latestNewView
	// directly return NewView response to help remote node recover view as we are in normal.
	rbft.logger.Infof("Replica %d sending new view response to %d, current view: %d, remote view: %d",
		rbft.chainConfig.SelfID, remoteReplicaID, rbft.chainConfig.View, remoteView)

	rcr := &consensus.RecoveryResponse{
		NewView:            nv,
		InitialCheckpoint:  rbft.storeMgr.localCheckpoints[rbft.chainConfig.H],
		GenesisBlockDigest: rbft.config.GenesisBlockDigest,
	}
	payload, err := rcr.MarshalVTStrict()
	if err != nil {
		rbft.logger.Errorf("ConsensusMessage_RECOVERY_RESPONSE Marshal Error: %s", err)
		return nil
	}
	consensusMsg := &consensus.ConsensusMessage{
		Type:    consensus.Type_RECOVERY_RESPONSE,
		Payload: payload,
	}
	rbft.peerMgr.unicast(context.TODO(), consensusMsg, remoteReplicaID)

	return nil
}

// recvRecoveryResponse process response.NewView to recover view and then process response.H to recover height.
func (rbft *rbftImpl[T, Constraint]) recvRecoveryResponse(rcr *consensus.RecoveryResponse) consensusEvent {
	if rcr.GetNewView() == nil || rcr.GetInitialCheckpoint() == nil {
		rbft.logger.Errorf("Replica %d received nil recovery response", rbft.chainConfig.SelfID)
		return nil
	}

	rbft.logger.Debugf("Replica %d received recovery response from %d", rbft.chainConfig.SelfID,
		rcr.GetInitialCheckpoint().GetAuthor())

	// current node is in recovery, which is caused by single node view change, cache this NewView
	// until received f+1 NewViews with the same targetView, and trigger single node recovery to
	// this targetView.
	if rbft.atomicIn(InRecovery) {
		if rcr.GenesisBlockDigest != rbft.config.GenesisBlockDigest {
			rbft.logger.Criticalf("Replica %d self genesis config is not consistent with most nodes, expected genesis block hash: %s, self genesis block hash: %s",
				rbft.chainConfig.SelfID, rcr.GenesisBlockDigest, rbft.config.GenesisBlockDigest)
		}
		nv := rcr.GetNewView()
		targetView, nvHash, valid := rbft.checkNewView(nv)
		if valid {
			// only process response in recovery status, which is caused by single node view change
			return rbft.resetStateForRecovery(targetView, nvHash, rcr)
		}
	}
	return nil
}

// primaryCheckNewView do some prepare for change to New view
// such as check if primary need state update and fetch missed batches
func (rbft *rbftImpl[T, Constraint]) primaryCheckNewView(xSet []*consensus.VcPq) consensusEvent {
	rbft.logger.Infof("New primary %d try to check new view", rbft.chainConfig.SelfID)

	// check if we have all request batch in xSet
	needFetchMissingReqBatch := rbft.checkIfNeedFetchMissingReqBatch(xSet)
	if needFetchMissingReqBatch {
		// try to fetch missing batches, if received all batches, jump into resetStateForNewView
		rbft.fetchRequestBatches()
		return nil
	}
	return rbft.resetStateForNewView()
}

// replicaCheckNewView checks this newView message and see if it's legal.
func (rbft *rbftImpl[T, Constraint]) replicaCheckNewView() consensusEvent {
	rbft.logger.Infof("Replica %d try to check new view", rbft.chainConfig.SelfID)

	nv, ok := rbft.vcMgr.newViewStore[rbft.chainConfig.View]
	if !ok {
		rbft.logger.Debugf("Replica %d ignore processNewView as it could not find view %d in its newViewStore",
			rbft.chainConfig.SelfID, rbft.chainConfig.View)
		return nil
	}

	if !rbft.atomicIn(InViewChange) {
		rbft.logger.Debugf("Replica %d reject newView as we are not in viewChange or recovery", rbft.chainConfig.SelfID)
		return nil
	}

	// avoid check new view again because of repeat ViewChangeQuorumEvent.
	if rbft.atomicIn(InViewChange) && rbft.vcMgr.vcHandled {
		rbft.logger.Debugf("Replica %d enter check new view in viewchange again, ignore it", rbft.chainConfig.SelfID)
		return nil
	}

	basis := make([]*consensus.VcBasis, 0, len(nv.ViewChangeSet.ViewChanges))
	for _, vc := range nv.ViewChangeSet.ViewChanges {
		basis = append(basis, vc.Basis)
	}

	checkpointState, checkpointSet, ok := rbft.selectInitialCheckpoint(basis)
	if !ok {
		rbft.logger.Infof("Replica %d could not determine initial checkpoint", rbft.chainConfig.SelfID)
		return rbft.sendViewChange()
	}
	rbft.logger.Debugf("initial checkpoint: %+v", checkpointState)

	// check if the xset sent by new primary is built correctly by the aset
	msgList := rbft.assignSequenceNumbers(basis, checkpointState.Meta.Height)
	if msgList == nil {
		rbft.logger.Infof("Replica %d could not assign sequence numbers: %+v",
			rbft.chainConfig.SelfID, rbft.vcMgr.viewChangeStore)
		return rbft.sendViewChange()
	}
	rbft.logger.Debugf("x-set: %+v", msgList)

	// first ensure xset length is equal.
	if len(msgList) != len(nv.Xset) {
		rbft.logger.Warningf("Replica %d failed to verify newView xset: computed length %d, "+
			"received length %d", rbft.chainConfig.SelfID, len(msgList), len(nv.Xset))
		return rbft.sendViewChange()
	}

	// then ensure xset content is equal.
	for i, msg := range msgList {
		if msg.SequenceNumber != nv.Xset[i].SequenceNumber || msg.BatchDigest != nv.Xset[i].BatchDigest {
			rbft.logger.Warningf("Replica %d failed to verify newView xset: computed %+v, received %+v",
				rbft.chainConfig.SelfID, msgList, nv.Xset)
			return rbft.sendViewChange()
		}
	}

	// after checked new view in viewchange, set vcHandled active to avoid check new view again.
	if rbft.atomicIn(InViewChange) {
		rbft.vcMgr.vcHandled = true
	}

	// check if replica need state update
	needStateUpdate := rbft.checkIfNeedStateUpdate(checkpointState, checkpointSet)
	if needStateUpdate || rbft.atomicIn(StateTransferring) {
		rbft.logger.Debugf("Replica %d needs to catch up in viewChange", rbft.chainConfig.SelfID)
		return nil
	}

	// replica checks if we have all request batch in xSet
	needFetchMissingReqBatch := rbft.checkIfNeedFetchMissingReqBatch(msgList)
	if needFetchMissingReqBatch {
		// try to fetch missing batches, if received all batches, jump into resetStateForNewView
		rbft.fetchRequestBatches()
		return nil
	}
	return rbft.resetStateForNewView()
}

// resetStateForNewView reset all states for new view
func (rbft *rbftImpl[T, Constraint]) resetStateForNewView() consensusEvent {
	nv, ok := rbft.vcMgr.newViewStore[rbft.chainConfig.View]
	if !ok || nv == nil {
		rbft.logger.Warningf("Replica %d ignore processReqInNewView as it could not find view %d in its "+
			"newViewStore", rbft.chainConfig.SelfID, rbft.chainConfig.View)
		return nil
	}

	if !rbft.atomicIn(InViewChange) {
		rbft.logger.Debugf("Replica %d is not in viewChange or recovery, not process new view", rbft.chainConfig.SelfID)
		return nil
	}

	rbft.logger.Debugf("Replica %d accept newView to view %d", rbft.chainConfig.SelfID, rbft.chainConfig.View)

	// empty the outstandingReqBatch, it is useless since new primary will resend pre-prepare
	rbft.cleanOutstandingAndCert()

	// set seqNo to lastExec for new primary to sort following batches from correct seqNo.
	rbft.batchMgr.setSeqNo(rbft.exec.lastExec)

	// clear requestPool cache to a correct state.
	rbft.putBackRequestBatches(nv.Xset)

	// clear consensus cache to a correct state.
	rbft.processNewView(nv.Xset)

	rbft.persistNewView(nv, false)
	rbft.logger.Infof("Replica %d persist view=%d after new view", rbft.chainConfig.SelfID, rbft.chainConfig.View)

	if rbft.atomicIn(InViewChange) {
		return &LocalEvent{
			Service:   ViewChangeService,
			EventType: ViewChangeDoneEvent,
		}
	}
	return nil
}

// resetStateForRecovery is triggered during single node recovery, cache this NewView until received
// f+1 NewViews with the same targetView, and trigger single node recovery to this targetView.
func (rbft *rbftImpl[T, Constraint]) resetStateForRecovery(targetView uint64, nvHash string, rcr *consensus.RecoveryResponse) consensusEvent {
	// verify correctness of initial checkpoint.
	initialCheckpoint := rcr.GetInitialCheckpoint()
	if initialCheckpoint == nil {
		return nil
	}
	// ignore NewView with no forward author.
	forwardPeer := initialCheckpoint.GetAuthor()
	if forwardPeer == 0 {
		return nil
	}
	vErr := rbft.verifySignedCheckpoint(initialCheckpoint)
	if vErr != nil {
		rbft.logger.Errorf("Replica %d verify signature of checkpoint from %d error: %s",
			rbft.chainConfig.SelfID, initialCheckpoint.GetAuthor(), vErr)
		return nil
	}
	initialCheckpointHeight := initialCheckpoint.GetCheckpoint().Height()
	initialCheckpointDigest := initialCheckpoint.GetCheckpoint().Digest()
	isConfig := initialCheckpoint.GetCheckpoint().NeedUpdateEpoch
	initialCheckpointState := types.CheckpointState{
		Meta: types.MetaState{
			Height: initialCheckpointHeight,
			Digest: initialCheckpointDigest,
		},
		IsConfig: isConfig,
	}
	// cache and wait f+1 same NewView if we are in recovery status.
	nvIdx := newViewIdx{
		targetView:             targetView,
		newViewHash:            nvHash,
		initialCheckpointState: initialCheckpointState,
	}
	nc, ok := rbft.vcMgr.newViewCache[nvIdx]
	if !ok {
		rbft.logger.Infof("Replica %d starts cache new view %d forward from %d with stable "+
			"checkpoint height %d, checkpoint digest: %s, checkpoint isConfig: %v, newViewHash: %s", rbft.chainConfig.SelfID, targetView, forwardPeer,
			initialCheckpointHeight,
			initialCheckpointDigest,
			isConfig,
			nvHash,
		)
		nc = &newViewCert{}
	}
	nc.forwardPeers = append(nc.forwardPeers, forwardPeer)
	nc.initialCheckpoints = append(nc.initialCheckpoints, initialCheckpoint)
	rbft.vcMgr.newViewCache[nvIdx] = nc
	count := len(nc.forwardPeers)
	rbft.logger.Infof("Replica %d cached %d new view %d, need %d", rbft.chainConfig.SelfID, count,
		targetView, rbft.oneCorrectQuorum())
	if count >= rbft.oneCorrectQuorum() {
		// check if replica need state update before check new view as initialCheckpointState may be newer
		// than state in new view.
		needStateUpdate := rbft.checkIfNeedStateUpdate(&initialCheckpointState, nc.initialCheckpoints)
		if needStateUpdate {
			rbft.logger.Noticef("Replica %d try to sync to initial checkpoint height %d, current h %d",
				rbft.chainConfig.SelfID, initialCheckpointHeight, rbft.chainConfig.H)
		}
		// close vcResendTimer
		rbft.timerMgr.stopTimer(vcResendTimer)
		rbft.logger.Noticef("Replica %d try to recovery view to %d", rbft.chainConfig.SelfID, targetView)
		rbft.setView(targetView)
		rbft.vcMgr.newViewStore[targetView] = rcr.NewView
		return rbft.replicaCheckNewView()
	}
	return nil
}

// used in view-change to fetch missing assigned, non-checkpointed requests
func (rbft *rbftImpl[T, Constraint]) fetchRequestBatches() {
	for digest := range rbft.storeMgr.missingReqBatches {
		rbft.logger.Debugf("Replica %d try to fetch missing request batch with digest: %s", rbft.chainConfig.SelfID, digest)
		frb := &consensus.FetchBatchRequest{
			BatchDigest: digest,
			ReplicaId:   rbft.chainConfig.SelfID,
		}
		payload, err := frb.MarshalVTStrict()
		if err != nil {
			rbft.logger.Errorf("ConsensusMessage_FetchBatchRequest Marshal Error: %s", err)
			return
		}
		consensusMsg := &consensus.ConsensusMessage{
			Type:    consensus.Type_FETCH_BATCH_REQUEST,
			Payload: payload,
		}
		rbft.metrics.fetchRequestBatchCounter.Add(float64(1))
		rbft.peerMgr.broadcast(context.TODO(), consensusMsg)
	}
}

// recvFetchRequestBatch returns the requested batch
func (rbft *rbftImpl[T, Constraint]) recvFetchRequestBatch(fr *consensus.FetchBatchRequest) error {
	rbft.logger.Debugf("Replica %d received fetch request batch from replica %d with digest: %s",
		rbft.chainConfig.SelfID, fr.ReplicaId, fr.BatchDigest)

	// check if we have requested batch
	digest := fr.BatchDigest
	if _, ok := rbft.storeMgr.batchStore[digest]; !ok {
		return nil // we don't have it either
	}

	rbft.logger.Debugf("Replica %d response request batch with digest: %s", rbft.chainConfig.SelfID, fr.BatchDigest)
	reqBatch := rbft.storeMgr.batchStore[digest]

	pbBatch, err := reqBatch.ToPB()
	if err != nil {
		rbft.logger.Errorf("RequestBatch marshal Error: %s", err)
		return nil
	}
	batch := &consensus.FetchBatchResponse{
		Batch:       pbBatch,
		BatchDigest: digest,
		ReplicaId:   rbft.chainConfig.SelfID,
	}
	payload, err := batch.MarshalVTStrict()
	if err != nil {
		rbft.logger.Errorf("ConsensusMessage_FetchBatchResponse Marshal Error: %s", err)
		return nil
	}
	consensusMsg := &consensus.ConsensusMessage{
		Type:    consensus.Type_FETCH_BATCH_RESPONSE,
		Payload: payload,
	}
	rbft.peerMgr.unicast(context.TODO(), consensusMsg, fr.ReplicaId)

	return nil
}

// recvFetchBatchResponse receives the FetchBatchResponse from other peers
// If receive all request batch, jump to resetStateForNewView.
func (rbft *rbftImpl[T, Constraint]) recvFetchBatchResponse(batch *consensus.FetchBatchResponse) consensusEvent {
	if batch == nil {
		rbft.logger.Errorf("Replica %d received request batch response with a nil batch", rbft.chainConfig.SelfID)
		return nil
	}

	rbft.logger.Debugf("Replica %d received missing request batch from replica %d with digest: %s",
		rbft.chainConfig.SelfID, batch.ReplicaId, batch.BatchDigest)

	digest := batch.BatchDigest
	if _, ok := rbft.storeMgr.missingReqBatches[digest]; !ok {
		rbft.logger.Debugf("Replica %d received missing request: %s, but we don't miss this request, ignore it",
			rbft.chainConfig.SelfID, digest)
		return nil // either the wrong digest, or we got it already from someone else
	}
	// store into batchStore only, and store into requestPool by order when processNewView.
	receiveBatch := &RequestBatch[T, Constraint]{}
	if err := receiveBatch.FromPB(batch.Batch); err != nil {
		rbft.logger.Errorf("RequestBatch unmarshal Error: %s", err)
		return nil
	}
	rbft.storeMgr.batchStore[digest] = receiveBatch
	rbft.persistBatch(digest)
	rbft.metrics.batchesGauge.Add(float64(1))

	// delete missingReqBatches in this batch
	delete(rbft.storeMgr.missingReqBatches, digest)

	// if receive all request batch, try to process new view
	if len(rbft.storeMgr.missingReqBatches) == 0 {
		if rbft.atomicIn(InViewChange) {
			_, ok := rbft.vcMgr.newViewStore[rbft.chainConfig.View]
			if !ok {
				rbft.logger.Warningf("Replica %d ignore resetStateForNewView as it could not find view %d in "+
					"its newViewStore", rbft.chainConfig.SelfID, rbft.chainConfig.View)
				return nil
			}
			return rbft.resetStateForNewView()
		}
	}
	return nil
}

// ##########################################################################
//           view change auxiliary functions
// ##########################################################################

// stopNewViewTimer stops newViewTimer
func (rbft *rbftImpl[T, Constraint]) stopNewViewTimer() {
	rbft.logger.Debugf("Replica %d stop a running newView timer", rbft.chainConfig.SelfID)
	rbft.timerMgr.stopTimer(newViewTimer)
}

// softstartNewViewTimer starts a new view timer no matter how many existed new view timer
func (rbft *rbftImpl[T, Constraint]) softStartNewViewTimer(timeout time.Duration, reason string, isNewView bool) {
	rbft.logger.Debugf("Replica %d soft start newView timer for %s: %s", rbft.chainConfig.SelfID, timeout, reason)

	event := &LocalEvent{
		Service:   ViewChangeService,
		EventType: ViewChangeTimerEvent,
	}
	// set nextDemandView to current view because we will wait for lastNewViewTimeout
	// to confirm if the primary(in view nextDemandView) can actually finish this round
	// of viewChange by sending newView in time, but we may concurrently
	// receive f+1 others' viewChange to nextDemandView+1 and lastNewViewTimeoutEvent, so
	// we need to ensure if we really need to send viewChange when receive lastNewViewTimeoutEvent.
	if isNewView {
		event.Event = nextDemandNewView(rbft.chainConfig.View)
	}

	hasStarted, _ := rbft.timerMgr.softStartTimerWithNewTT(newViewTimer, timeout, event)
	if hasStarted {
		rbft.logger.Debugf("Replica %d has started new view timer before", rbft.chainConfig.SelfID)
	} else {
		rbft.vcMgr.newViewTimerReason = reason
	}
}

// beforeSendVC operates before send view change
// 1. Stop NewViewTimer and nullRequestTimer
// 2. increase the view and delete new view of old view in newViewStore
// 3. update pqlist
// 4. delete old viewChange message
func (rbft *rbftImpl[T, Constraint]) beforeSendVC(recovery bool) error {
	rbft.stopNewViewTimer()
	rbft.timerMgr.stopTimer(nullRequestTimer)
	rbft.stopFetchCheckpointTimer()
	rbft.stopHighWatermarkTimer()

	rbft.atomicOn(InViewChange)
	rbft.metrics.statusGaugeInViewChange.Set(InViewChange)
	rbft.setAbNormal()
	rbft.vcMgr.vcHandled = false

	// as we try to view-change, current node should close config-change status as we will
	// reach a correct state after view-change
	rbft.epochMgr.configBatchToCheck = nil
	rbft.atomicOff(InConfChange)
	rbft.epochMgr.configBatchInOrder = 0
	rbft.metrics.statusGaugeInConfChange.Set(0)

	newView := rbft.chainConfig.View + uint64(1)
	if recovery {
		rbft.setViewWithRecovery(newView)
	} else {
		rbft.setView(newView)
	}

	delete(rbft.vcMgr.newViewStore, rbft.chainConfig.View)

	// clear old messages
	for idx := range rbft.vcMgr.viewChangeStore {
		if idx.v < rbft.chainConfig.View {
			delete(rbft.vcMgr.viewChangeStore, idx)
		}
	}
	return nil
}

// correctViewChange checks if view change basis correct
// 1. pqlist' view should be less than vc.View and SequenceNumber should be larger than vc.H.
// 2. checkpoint's SequenceNumber should be larger than vc.H
func (rbft *rbftImpl[T, Constraint]) correctViewChange(vcBasis *consensus.VcBasis) bool {
	for _, p := range append(vcBasis.GetPset(), vcBasis.GetQset()...) {
		if !(p.View < vcBasis.GetView() && p.SequenceNumber > vcBasis.GetH()) {
			rbft.logger.Debugf("Replica %d find invalid p entry in viewChange: vcBasis(v:%d h:%d) p(v:%d n:%d)",
				rbft.chainConfig.SelfID, vcBasis.GetView(), vcBasis.GetH(), p.View, p.SequenceNumber)
			return false
		}
	}

	for _, c := range vcBasis.GetCset() {
		if !(c.GetCheckpoint().Height() >= vcBasis.GetH()) {
			rbft.logger.Debugf("Replica %d find invalid c entry in viewChange: vcBasis(v:%d h:%d) c(n:%d d:%s)",
				rbft.chainConfig.SelfID, vcBasis.GetView(), vcBasis.GetH(), c.GetCheckpoint().Height(), c.GetCheckpoint().GetExecuteState().Digest)
			return false
		}
	}

	return true
}

// getViewChangeBasis returns all viewChange basis from viewChangeStore
func (rbft *rbftImpl[T, Constraint]) getViewChangeBasis() (*consensus.QuorumViewChange, []*consensus.VcBasis) {
	qvc := &consensus.QuorumViewChange{
		ViewChanges: make([]*consensus.ViewChange, 0, len(rbft.vcMgr.viewChangeStore)),
	}
	basis := make([]*consensus.VcBasis, 0, len(rbft.vcMgr.viewChangeStore))
	for idx, vc := range rbft.vcMgr.viewChangeStore {
		if idx.v == rbft.chainConfig.View {
			qvc.ViewChanges = append(qvc.ViewChanges, vc)
			basis = append(basis, vc.Basis)
		}
	}
	return qvc, basis
}

type ValidatorDynamicInfo struct {
	count int
	info  []*consensus.NodeDynamicInfo
	hash  string
}

func (v *ValidatorDynamicInfo) getHash() (string, error) {
	if v.hash != "" {
		return v.hash, nil
	}
	hasher := sha3.NewLegacyKeccak256()
	pbInfo := &consensus.ValidatorDynamicInfo{
		Info: v.info,
	}
	raw, err := pbInfo.MarshalVTStrict()
	if err != nil {
		return "", err
	}
	_, err = hasher.Write(raw)
	if err != nil {
		return "", err
	}
	hash := hasher.Sum(nil)
	v.hash = hex.EncodeToString(hash)
	return v.hash, nil
}

func (rbft *rbftImpl[T, Constraint]) selectValidatorDynamicInfo(viewChanges []*consensus.ViewChange) ([]*consensus.NodeDynamicInfo, bool) {
	recoverValidatorDynamicInfoRecord := make(map[string]*ValidatorDynamicInfo)
	notRecoverValidatorDynamicInfoRecord := make(map[string]*ValidatorDynamicInfo)
	var recoverCount, notRecoverCount int
	var quorumRecoverValidatorDynamicInfo, quorumNotRecoverValidatorDynamicInfo []*consensus.NodeDynamicInfo
	for _, vc := range viewChanges {
		b := vc.Basis
		if len(b.IfRecoverValidatorDynamicInfo) == 0 || len(b.IfNotRecoverValidatorDynamicInfo) == 0 {
			continue
		}

		recoverValidatorDynamicInfo := &ValidatorDynamicInfo{
			count: 1,
			info:  b.IfRecoverValidatorDynamicInfo,
		}
		hash, err := recoverValidatorDynamicInfo.getHash()
		if err != nil {
			continue
		}
		if info, ok := recoverValidatorDynamicInfoRecord[hash]; ok {
			info.count++
			if info.count == rbft.commonCaseQuorum() {
				quorumRecoverValidatorDynamicInfo = recoverValidatorDynamicInfo.info
			}
		} else {
			recoverValidatorDynamicInfoRecord[hash] = recoverValidatorDynamicInfo
		}

		notRecoverValidatorDynamicInfo := &ValidatorDynamicInfo{
			count: 1,
			info:  b.IfNotRecoverValidatorDynamicInfo,
		}
		hash, err = notRecoverValidatorDynamicInfo.getHash()
		if err != nil {
			continue
		}
		if info, ok := notRecoverValidatorDynamicInfoRecord[hash]; ok {
			info.count++
			if info.count == rbft.commonCaseQuorum() {
				quorumNotRecoverValidatorDynamicInfo = notRecoverValidatorDynamicInfo.info
			}
		} else {
			notRecoverValidatorDynamicInfoRecord[hash] = notRecoverValidatorDynamicInfo
		}

		if vc.Recovery {
			recoverCount++
		} else {
			notRecoverCount++
		}
	}
	if len(quorumRecoverValidatorDynamicInfo) == 0 && len(quorumNotRecoverValidatorDynamicInfo) == 0 {
		return nil, false
	} else if len(quorumRecoverValidatorDynamicInfo) != 0 && len(quorumNotRecoverValidatorDynamicInfo) == 0 {
		return quorumRecoverValidatorDynamicInfo, true
	} else if len(quorumRecoverValidatorDynamicInfo) == 0 && len(quorumNotRecoverValidatorDynamicInfo) != 0 {
		return quorumNotRecoverValidatorDynamicInfo, true
	}
	if recoverCount >= notRecoverCount {
		return quorumRecoverValidatorDynamicInfo, true
	}
	return quorumNotRecoverValidatorDynamicInfo, true
}

// selectInitialCheckpoint selects suitable initial checkpoint from received ViewChange message.
// If we find suitable checkpoint, it returns a certain checkpoint meta and the signed checkpoint set.
// The initial checkpoint is the max checkpoint which exists in at least oneCorrectQuorum
// peers and greater than low waterMark of at least commonCaseQuorum.
func (rbft *rbftImpl[T, Constraint]) selectInitialCheckpoint(set []*consensus.VcBasis) (*types.CheckpointState, []*consensus.SignedCheckpoint, bool) {
	var (
		initialCheckpointState = types.CheckpointState{}
		checkpointSet          []*consensus.SignedCheckpoint
		find                   bool
	)

	// for the checkpoint as key, find the corresponding basis messages
	checkpoints := make(map[types.CheckpointState][]*consensus.SignedCheckpoint)
	for _, basis := range set {
		// verify that we strip duplicate checkpoints from this Cset
		record := make(map[types.CheckpointState]bool)
		for _, c := range basis.GetCset() {
			if c == nil || c.GetCheckpoint() == nil {
				rbft.logger.Warningf("Replica %d received an invalid vc basis with nil checkpoint", rbft.chainConfig.SelfID)
				continue
			}
			height := c.GetCheckpoint().Height()
			digest := c.GetCheckpoint().Digest()
			cs := types.CheckpointState{Meta: types.MetaState{Height: height, Digest: digest},
				IsConfig: c.GetCheckpoint().NeedUpdateEpoch}
			if ok := record[cs]; ok {
				continue
			}
			checkpoints[cs] = append(checkpoints[cs], c)
			record[cs] = true
			rbft.logger.Debugf("Replica %d appending checkpoint from replica %d with "+
				"seqNo=%d, h=%d, and checkpoint digest %s", rbft.chainConfig.SelfID, basis.GetReplicaId(), height, basis.GetH(), digest)
		}
	}

	// indicate that replica cannot find any weak checkpoint cert
	if len(checkpoints) == 0 {
		rbft.logger.Debugf("Replica %d has no checkpoints to select from: %d %s",
			rbft.chainConfig.SelfID, len(rbft.vcMgr.viewChangeStore), checkpoints)
		return nil, nil, false
	}

	for chkptIdx, signedCheckpoints := range checkpoints {
		// need weak certificate for the checkpoint
		if len(signedCheckpoints) < rbft.oneCorrectQuorum() {
			rbft.logger.Debugf("Replica %d has no weak certificate for n:%d, signedCheckpoints was %d long",
				rbft.chainConfig.SelfID, chkptIdx.Meta.Height, len(signedCheckpoints))
			continue
		}

		// for config checkpoint, we need at least quorum signatures as this config checkpoint
		// will be recorded on ledger for the proof base.
		if chkptIdx.IsConfig && len(signedCheckpoints) < rbft.commonCaseQuorum() {
			rbft.logger.Warningf("Replica %d has no quorum for n:%d, config signedCheckpoints was %d long",
				rbft.chainConfig.SelfID, chkptIdx.Meta.Height, len(signedCheckpoints))
			continue
		}

		quorum := 0
		// Note, this is the whole x-set (S) in the paper, not just this checkpoint set (S') (signedCheckpoints)
		// We need 2f+1 low watermarks from S below this seqNo from all replicas
		// We need f+1 matching checkpoints at this seqNo (S')
		for _, basis := range set {
			if basis.GetH() <= chkptIdx.Meta.Height {
				quorum++
			}
		}

		if quorum < rbft.commonCaseQuorum() {
			rbft.logger.Debugf("Replica %d has no quorum for n:%d", rbft.chainConfig.SelfID, chkptIdx.Meta.Height)
			continue
		}

		// Find the highest checkpoint
		if initialCheckpointState.Meta.Height <= chkptIdx.Meta.Height {
			initialCheckpointState = chkptIdx
			checkpointSet = signedCheckpoints
			find = true
		}
	}

	if !find {
		return nil, nil, false
	}

	validCheckpoints := make([]*consensus.SignedCheckpoint, 0, len(checkpointSet))
	for _, signedCheckpoint := range checkpointSet {
		err := rbft.verifySignedCheckpoint(signedCheckpoint)
		if err != nil {
			rbft.logger.Errorf("Replica %d verify signature of checkpoint %s from %d error: %s",
				rbft.chainConfig.SelfID, signedCheckpoint, signedCheckpoint.GetAuthor(), err)
		} else {
			validCheckpoints = append(validCheckpoints, signedCheckpoint)
		}
	}
	if len(validCheckpoints) < rbft.oneCorrectQuorum() {
		rbft.logger.Debugf("Replica %d has no valid weak certificate for n:%d, signedCheckpoints was %d long",
			rbft.chainConfig.SelfID, initialCheckpointState.Meta.Height, len(validCheckpoints))
		return nil, nil, false
	}

	if initialCheckpointState.Meta.Height == rbft.config.GenesisEpochInfo.StartBlock {
		if initialCheckpointState.Meta.Digest != rbft.config.GenesisBlockDigest {
			rbft.logger.Criticalf("Replica %d self genesis config is not consistent with most nodes, expected genesis block hash: %s, self genesis block hash: %s",
				rbft.chainConfig.SelfID, initialCheckpointState.Meta.Digest, rbft.config.GenesisBlockDigest)
		}
	}

	return &initialCheckpointState, checkpointSet, true
}

// assignSequenceNumbers selects a request to pre-prepare in the new view
// for each sequence number n between h and h + L, which is according to
// Castro's TOCS PBFT, Fig. 4.
func (rbft *rbftImpl[T, Constraint]) assignSequenceNumbers(set []*consensus.VcBasis, h uint64) []*consensus.VcPq {
	msgMap := make(map[uint64]string)

	maxN := h + 1

	// "for all n such that h < n <= h + L"
nLoop:
	for n := h + 1; n <= h+rbft.chainConfig.L; n++ {
		// "∃m ∈ S..."
		for _, m := range set {
			// "...with <n,d,v> ∈ m.P"
			for _, em := range m.GetPset() {
				if n != em.SequenceNumber {
					continue
				}
				quorum := 0
				// "A1. ∃2f+1 messages m' ∈ S"
			mpLoop:
				for _, mp := range set {
					if mp.GetH() >= n {
						continue
					}
					// "∀<n,d',v'> ∈ m'.P"
					for _, emp := range mp.GetPset() {
						if n != emp.SequenceNumber {
							continue
						}
						if !(emp.View < em.View || (emp.View == em.View && emp.BatchDigest == em.BatchDigest)) {
							continue mpLoop
						}
					}
					quorum++
				}

				if quorum < rbft.commonCaseQuorum() {
					continue
				}

				quorum = 0
				// "A2. ∃f+1 messages m' ∈ S"
				for _, mp := range set {
					// "∃<n,d',v'> ∈ m'.Q"
					for _, emp := range mp.GetQset() {
						if n != emp.SequenceNumber {
							continue
						}
						if emp.View >= em.View && emp.BatchDigest == em.BatchDigest {
							quorum++
							break
						}
					}
				}

				if quorum < rbft.oneCorrectQuorum() {
					continue
				}

				// "then select the request with digest d for number n"
				msgMap[n] = em.BatchDigest
				maxN = n

				continue nLoop
			}
		}

		quorum := 0
		// "else if ∃2f+1 messages m ∈ S"
	nullLoop:
		for _, m := range set {
			// "m.h < n"
			if m.GetH() >= n {
				continue
			}
			// "m.P has no entry for n"
			for _, em := range m.GetPset() {
				if em.SequenceNumber == n {
					continue nullLoop
				}
			}
			quorum++
		}

		if quorum >= rbft.commonCaseQuorum() {
			// "then select the null request for number n"
			msgMap[n] = ""

			continue nLoop
		}

		rbft.logger.Warningf("Replica %d could not assign value to contents of seqNo %d, found only %d "+
			"missing P entries", rbft.chainConfig.SelfID, n, quorum)
		return nil
	}

	// prune top null requests
	// TODO(DH): is it safe to prune all null request beyond maxN?
	// if new primary update its seqNo to larger maxN?
	for n, msg := range msgMap {
		if n >= maxN && msg == "" {
			delete(msgMap, n)
		}
	}

	x := h + 1
	msgList := make([]*consensus.VcPq, 0, len(msgMap))
	for range msgMap {
		msgList = append(msgList, &consensus.VcPq{
			SequenceNumber: x,
			BatchDigest:    msgMap[x],
		})
		x++
	}

	return msgList
}

// updateViewChangeSeqNo updates viewChangeSeqNo by viewChangePeriod
func (rbft *rbftImpl[T, Constraint]) updateViewChangeSeqNo(seqNo, K uint64) {
	if rbft.vcMgr.viewChangePeriod <= 0 {
		return
	}
	// ensure the view change always occurs at a checkpoint boundary
	rbft.vcMgr.viewChangeSeqNo = seqNo - seqNo%K + rbft.vcMgr.viewChangePeriod*K
}

// checkIfNeedFetchMissingReqBatch checks if we need fetch missing reqBatch when this node
// doesn't have all reqBatch in xset.
func (rbft *rbftImpl[T, Constraint]) checkIfNeedFetchMissingReqBatch(xset []*consensus.VcPq) (newReqBatchMissing bool) {
	// clear missingReqBatches to ensure it's only valid in one recovery round.
	rbft.storeMgr.missingReqBatches = make(map[string]bool)
	newReqBatchMissing = false
	for _, msg := range xset {
		n := msg.SequenceNumber
		d := msg.BatchDigest
		// RBFT: why should we use "h ≥ min{n | ∃d : (<n,d> ∈ X)}"?
		// "h ≥ min{n | ∃d : (<n,d> ∈ X)} ∧ ∀<n,d> ∈ X : (n ≤ h ∨ ∃m ∈ in : (D(m) = d))"
		if n <= rbft.chainConfig.H {
			continue
		} else {
			if d == "" {
				// don't need to fetch null request.
				continue
			}

			if _, ok := rbft.storeMgr.batchStore[d]; !ok {
				rbft.logger.Debugf("Replica %d missing assigned, non-checkpointed request batch %s",
					rbft.chainConfig.SelfID, d)
				if _, missing := rbft.storeMgr.missingReqBatches[d]; !missing {
					rbft.logger.Infof("Replica %v needs to fetch batch %s", rbft.chainConfig.SelfID, d)
					newReqBatchMissing = true
					rbft.storeMgr.missingReqBatches[d] = true
				}
			}
		}
	}
	return newReqBatchMissing
}

// processNewView re-construct certStore using prePrepare and prepare with digest in xSet.
func (rbft *rbftImpl[T, Constraint]) processNewView(msgList []*consensus.VcPq) {
	if len(msgList) == 0 {
		rbft.logger.Debugf("Replica %d directly finish process new view as msgList is empty.", rbft.chainConfig.SelfID)
		return
	}

	isPrimary := rbft.isPrimary(rbft.chainConfig.SelfID)
	maxN := rbft.exec.lastExec

	for _, msg := range msgList {
		n := msg.SequenceNumber
		d := msg.BatchDigest

		if n <= rbft.chainConfig.H {
			rbft.logger.Debugf("Replica %d not process seqNo %d in view %d which is lower than h: %d",
				rbft.chainConfig.SelfID, n, rbft.chainConfig.View, rbft.chainConfig.H)
			continue
		}

		// check if we are lack of the txBatch with given digest
		// this should not happen as we must have fetched missing batch before we enter processNewView
		batch, ok := rbft.storeMgr.batchStore[d]
		if !ok && d != "" {
			rbft.logger.Warningf("Replica %d is missing tx batch for seqNo=%d with digest '%s' for "+
				"assigned seqNo", rbft.chainConfig.SelfID, n, d)
			continue
		}

		cert := rbft.storeMgr.getCert(rbft.chainConfig.View, n, d)

		prePrep := &consensus.PrePrepare{
			View:           rbft.chainConfig.View,
			SequenceNumber: n,
			BatchDigest:    d,
			ReplicaId:      rbft.chainConfig.PrimaryID,
		}
		if d == "" {
			rbft.logger.Infof("Replica %d need to process seqNo %d as a null request, proposer:%d", rbft.chainConfig.SelfID, n, rbft.chainConfig.PrimaryID)
			// construct prePrepare with an empty batch
			prePrep.HashBatch = &consensus.HashBatch{
				RequestHashList: []string{},
				Proposer:        rbft.chainConfig.PrimaryID,
			}
		} else {
			// put un-executed batch into outstandingReqBatches, if replica cannot execute this batch
			// during that timeout, this replica which will trigger viewChange.
			if n > rbft.exec.lastExec {
				rbft.storeMgr.outstandingReqBatches[d] = batch
			}

			// rebuild prePrepare with batch recorded in batchStore
			prePrep.HashBatch = &consensus.HashBatch{
				RequestHashList: batch.RequestHashList,
				Timestamp:       batch.Timestamp,
				Proposer:        batch.Proposer,
			}

			// re-construct batches by order in xSet to de-duplicate txs during different batches in msgList which
			// may be caused by different primary puts the same txs into different batches with different seqNo'
			oldBatch := &txpool.RequestHashBatch[T, Constraint]{
				BatchHash:  batch.BatchHash,
				TxHashList: batch.RequestHashList,
				TxList:     batch.RequestList,
				Timestamp:  batch.Timestamp,
			}
			deDuplicateTxHashes, err := rbft.batchMgr.requestPool.ReConstructBatchByOrder(oldBatch)
			if err != nil {
				rbft.logger.Warningf("Replica %d failed to re-construct batch %s, err: %s, send viewChange",
					rbft.chainConfig.SelfID, d, err)
				rbft.sendViewChange()
				return
			}
			if len(deDuplicateTxHashes) != 0 {
				rbft.logger.Noticef("Replica %d finds %d duplicate txs when re-construct batch %d with digest %s, "+
					"detailed: %+v", rbft.chainConfig.SelfID, len(deDuplicateTxHashes), n, d, deDuplicateTxHashes)
				prePrep.HashBatch.DeDuplicateRequestHashList = deDuplicateTxHashes
			}
		}
		cert.prePrepare = prePrep
		cert.prePrepareCtx = context.Background()
		rbft.persistQSet(prePrep)
		rbft.storeMgr.seqMap[n] = d
		if metrics.EnableExpensive() {
			cert.prePreparedTime = time.Now().UnixNano()
			duration := time.Duration(cert.prePreparedTime - prePrep.HashBatch.Timestamp).Seconds()
			rbft.metrics.batchToPrePrepared.Observe(duration)
		}

		if n > maxN {
			maxN = n
		}

		if isConfigBatch(n, rbft.chainConfig.EpochInfo) {
			rbft.logger.Noticef("Replica %d is processing a config batch %d, skip the following", rbft.chainConfig.SelfID, n)
			if n == rbft.exec.lastExec {
				rbft.atomicOn(InConfChange)
				cert.isConfig = true
				rbft.epochMgr.configBatchInOrder = n
				rbft.metrics.statusGaugeInConfChange.Set(InConfChange)
				// we have executed a config batch in old view, but we haven't reached stable checkpoint
				// for that batch, so we need to fetch the missing config checkpoint.
				state := rbft.node.getCurrentState()
				if n == state.MetaState.Height {
					rbft.epochMgr.configBatchToCheck = state.MetaState
					rbft.logger.Infof("Replica %d try to fetch config checkpoint", rbft.chainConfig.SelfID)
					rbft.fetchCheckpoint()
				} else {
					rbft.logger.Errorf("Replica %d has an incorrect state: %+v", rbft.chainConfig.SelfID, state)
				}
			} else if n > rbft.exec.lastExec {
				// we have batched but not executed the config batch, turn into ConfChange status and
				// execute this config batch later.
				rbft.atomicOn(InConfChange)
				cert.isConfig = true
				rbft.epochMgr.configBatchInOrder = n
				rbft.metrics.statusGaugeInConfChange.Set(InConfChange)
				rbft.logger.Infof("Replica %d finds config batch in x-set", rbft.chainConfig.SelfID)
			} else {
				rbft.logger.Warningf("Replica %d finds config batch %d in x-set, which is lower than lastExec %d",
					rbft.chainConfig.SelfID, n, rbft.exec.lastExec)
			}
			break
		}
	}
	// update seqNo as new primary needs to start prePrepare with a correct number.
	// NOTE: directly set seqNo to maxN in xSet.
	rbft.batchMgr.setSeqNo(maxN)

	for _, msg := range msgList {
		n := msg.SequenceNumber
		d := msg.BatchDigest
		// only backup needs to rebuild self's Prepare and broadcast this Prepare
		if !isPrimary {
			rbft.logger.Debugf("Replica %d sending prepare for view=%d/seqNo=%d/digest=%s after new view",
				rbft.chainConfig.SelfID, rbft.chainConfig.View, n, d)
			prep := &consensus.Prepare{
				ReplicaId:      rbft.chainConfig.SelfID,
				View:           rbft.chainConfig.View,
				SequenceNumber: n,
				BatchDigest:    d,
			}
			if n > rbft.chainConfig.H {
				cert := rbft.storeMgr.getCert(rbft.chainConfig.View, n, d)
				cert.sentPrepare = true
				_ = rbft.recvPrepare(context.TODO(), prep)
			}
			payload, err := prep.MarshalVTStrict()
			if err != nil {
				rbft.logger.Errorf("ConsensusMessage_PREPARE Marshal Error: %s", err)
				return
			}

			consensusMsg := &consensus.ConsensusMessage{
				Type:    consensus.Type_PREPARE,
				Payload: payload,
			}
			rbft.peerMgr.broadcast(context.TODO(), consensusMsg)
		}

		// directly construct commit message for committed batches even though we have not went through
		// prePrepare and Prepare phase in new view because we may lose commit message in the following
		// normal case(because of elimination rule of PQC), after which, if new node needs to fetchPQC
		// to recover state after stable checkpoint, it will not get enough commit messages to recover
		// to the latest height.
		// NOTE: this is always correct to construct certs of committed batches.
		if n > rbft.chainConfig.H && n <= rbft.exec.lastExec {
			rbft.logger.Debugf("Replica %d sending commit for view=%d/seqNo=%d/digest=%s after new view",
				rbft.chainConfig.SelfID, rbft.chainConfig.View, n, d)
			cmt := &consensus.Commit{
				ReplicaId:      rbft.chainConfig.SelfID,
				View:           rbft.chainConfig.View,
				SequenceNumber: n,
				BatchDigest:    d,
			}

			cert := rbft.storeMgr.getCert(rbft.chainConfig.View, n, d)
			cert.sentCommit = true
			_ = rbft.recvCommit(context.TODO(), cmt)

			payload, err := cmt.MarshalVTStrict()
			if err != nil {
				rbft.logger.Errorf("ConsensusMessage_COMMIT Marshal Error: %s", err)
				return
			}

			consensusMsg := &consensus.ConsensusMessage{
				Type:    consensus.Type_COMMIT,
				Payload: payload,
			}
			rbft.peerMgr.broadcast(context.TODO(), consensusMsg)
		}
	}
}

func showSimpleValidatorDynamicInfo(v []*consensus.NodeDynamicInfo) []ValidatorInfo {
	return lo.Map(v, func(item *consensus.NodeDynamicInfo, index int) ValidatorInfo {
		return ValidatorInfo{
			ID:                             item.Id,
			ConsensusVotingPower:           item.ConsensusVotingPower,
			ConsensusVotingPowerReduced:    item.ConsensusVotingPowerReduced,
			ConsensusVotingPowerReduceView: item.ConsensusVotingPowerReduceView,
		}
	})
}