|
|
- package consensus
-
- import (
- "bytes"
- "errors"
- "fmt"
- "path"
- "reflect"
- "sync"
- "time"
-
- fail "github.com/ebuchman/fail-test"
- wire "github.com/tendermint/go-wire"
- cfg "github.com/tendermint/tendermint/config"
- "github.com/tendermint/tendermint/proxy"
- sm "github.com/tendermint/tendermint/state"
- "github.com/tendermint/tendermint/types"
- cmn "github.com/tendermint/tmlibs/common"
- "github.com/tendermint/tmlibs/log"
- )
-
- //-----------------------------------------------------------------------------
- // Config
-
- //-----------------------------------------------------------------------------
- // Errors
-
- var (
- ErrInvalidProposalSignature = errors.New("Error invalid proposal signature")
- ErrInvalidProposalPOLRound = errors.New("Error invalid proposal POL round")
- ErrAddingVote = errors.New("Error adding vote")
- ErrVoteHeightMismatch = errors.New("Error vote height mismatch")
- )
-
- //-----------------------------------------------------------------------------
- // RoundStepType enum type
-
- // RoundStepType enumerates the state of the consensus state machine
- type RoundStepType uint8 // These must be numeric, ordered.
-
- const (
- RoundStepNewHeight = RoundStepType(0x01) // Wait til CommitTime + timeoutCommit
- RoundStepNewRound = RoundStepType(0x02) // Setup new round and go to RoundStepPropose
- RoundStepPropose = RoundStepType(0x03) // Did propose, gossip proposal
- RoundStepPrevote = RoundStepType(0x04) // Did prevote, gossip prevotes
- RoundStepPrevoteWait = RoundStepType(0x05) // Did receive any +2/3 prevotes, start timeout
- RoundStepPrecommit = RoundStepType(0x06) // Did precommit, gossip precommits
- RoundStepPrecommitWait = RoundStepType(0x07) // Did receive any +2/3 precommits, start timeout
- RoundStepCommit = RoundStepType(0x08) // Entered commit state machine
- // NOTE: RoundStepNewHeight acts as RoundStepCommitWait.
- )
-
- // String returns a string
- func (rs RoundStepType) String() string {
- switch rs {
- case RoundStepNewHeight:
- return "RoundStepNewHeight"
- case RoundStepNewRound:
- return "RoundStepNewRound"
- case RoundStepPropose:
- return "RoundStepPropose"
- case RoundStepPrevote:
- return "RoundStepPrevote"
- case RoundStepPrevoteWait:
- return "RoundStepPrevoteWait"
- case RoundStepPrecommit:
- return "RoundStepPrecommit"
- case RoundStepPrecommitWait:
- return "RoundStepPrecommitWait"
- case RoundStepCommit:
- return "RoundStepCommit"
- default:
- return "RoundStepUnknown" // Cannot panic.
- }
- }
-
- //-----------------------------------------------------------------------------
-
- // RoundState defines the internal consensus state.
- // It is Immutable when returned from ConsensusState.GetRoundState()
- // TODO: Actually, only the top pointer is copied,
- // so access to field pointers is still racey
- type RoundState struct {
- Height int // Height we are working on
- Round int
- Step RoundStepType
- StartTime time.Time
- CommitTime time.Time // Subjective time when +2/3 precommits for Block at Round were found
- Validators *types.ValidatorSet
- Proposal *types.Proposal
- ProposalBlock *types.Block
- ProposalBlockParts *types.PartSet
- LockedRound int
- LockedBlock *types.Block
- LockedBlockParts *types.PartSet
- Votes *HeightVoteSet
- CommitRound int //
- LastCommit *types.VoteSet // Last precommits at Height-1
- LastValidators *types.ValidatorSet
- }
-
- // RoundStateEvent returns the H/R/S of the RoundState as an event.
- func (rs *RoundState) RoundStateEvent() types.EventDataRoundState {
- edrs := types.EventDataRoundState{
- Height: rs.Height,
- Round: rs.Round,
- Step: rs.Step.String(),
- RoundState: rs,
- }
- return edrs
- }
-
- // String returns a string
- func (rs *RoundState) String() string {
- return rs.StringIndented("")
- }
-
- // StringIndented returns a string
- func (rs *RoundState) StringIndented(indent string) string {
- return fmt.Sprintf(`RoundState{
- %s H:%v R:%v S:%v
- %s StartTime: %v
- %s CommitTime: %v
- %s Validators: %v
- %s Proposal: %v
- %s ProposalBlock: %v %v
- %s LockedRound: %v
- %s LockedBlock: %v %v
- %s Votes: %v
- %s LastCommit: %v
- %s LastValidators: %v
- %s}`,
- indent, rs.Height, rs.Round, rs.Step,
- indent, rs.StartTime,
- indent, rs.CommitTime,
- indent, rs.Validators.StringIndented(indent+" "),
- indent, rs.Proposal,
- indent, rs.ProposalBlockParts.StringShort(), rs.ProposalBlock.StringShort(),
- indent, rs.LockedRound,
- indent, rs.LockedBlockParts.StringShort(), rs.LockedBlock.StringShort(),
- indent, rs.Votes.StringIndented(indent+" "),
- indent, rs.LastCommit.StringShort(),
- indent, rs.LastValidators.StringIndented(indent+" "),
- indent)
- }
-
- // StringShort returns a string
- func (rs *RoundState) StringShort() string {
- return fmt.Sprintf(`RoundState{H:%v R:%v S:%v ST:%v}`,
- rs.Height, rs.Round, rs.Step, rs.StartTime)
- }
-
- //-----------------------------------------------------------------------------
-
- var (
- msgQueueSize = 1000
- )
-
- // msgs from the reactor which may update the state
- type msgInfo struct {
- Msg ConsensusMessage `json:"msg"`
- PeerKey string `json:"peer_key"`
- }
-
- // internally generated messages which may update the state
- type timeoutInfo struct {
- Duration time.Duration `json:"duration"`
- Height int `json:"height"`
- Round int `json:"round"`
- Step RoundStepType `json:"step"`
- }
-
- func (ti *timeoutInfo) String() string {
- return fmt.Sprintf("%v ; %d/%d %v", ti.Duration, ti.Height, ti.Round, ti.Step)
- }
-
- // PrivValidator is a validator that can sign votes and proposals.
- type PrivValidator interface {
- GetAddress() []byte
- SignVote(chainID string, vote *types.Vote) error
- SignProposal(chainID string, proposal *types.Proposal) error
- }
-
- // ConsensusState handles execution of the consensus algorithm.
- // It processes votes and proposals, and upon reaching agreement,
- // commits blocks to the chain and executes them against the application.
- // The internal state machine receives input from peers, the internal validator, and from a timer.
- type ConsensusState struct {
- cmn.BaseService
-
- // config details
- config *cfg.ConsensusConfig
- privValidator PrivValidator // for signing votes
-
- // services for creating and executing blocks
- proxyAppConn proxy.AppConnConsensus
- blockStore types.BlockStore
- mempool types.Mempool
-
- // internal state
- mtx sync.Mutex
- RoundState
- state *sm.State // State until height-1.
-
- // state changes may be triggered by msgs from peers,
- // msgs from ourself, or by timeouts
- peerMsgQueue chan msgInfo
- internalMsgQueue chan msgInfo
- timeoutTicker TimeoutTicker
-
- // we use PubSub to trigger msg broadcasts in the reactor,
- // and to notify external subscribers, eg. through a websocket
- evsw types.EventSwitch
-
- // a Write-Ahead Log ensures we can recover from any kind of crash
- // and helps us avoid signing conflicting votes
- wal *WAL
- replayMode bool // so we don't log signing errors during replay
-
- // for tests where we want to limit the number of transitions the state makes
- nSteps int
-
- // some functions can be overwritten for testing
- decideProposal func(height, round int)
- doPrevote func(height, round int)
- setProposal func(proposal *types.Proposal) error
-
- // closed when we finish shutting down
- done chan struct{}
- }
-
- // NewConsensusState returns a new ConsensusState.
- func NewConsensusState(config *cfg.ConsensusConfig, state *sm.State, proxyAppConn proxy.AppConnConsensus, blockStore types.BlockStore, mempool types.Mempool) *ConsensusState {
- cs := &ConsensusState{
- config: config,
- proxyAppConn: proxyAppConn,
- blockStore: blockStore,
- mempool: mempool,
- peerMsgQueue: make(chan msgInfo, msgQueueSize),
- internalMsgQueue: make(chan msgInfo, msgQueueSize),
- timeoutTicker: NewTimeoutTicker(),
- done: make(chan struct{}),
- }
- // set function defaults (may be overwritten before calling Start)
- cs.decideProposal = cs.defaultDecideProposal
- cs.doPrevote = cs.defaultDoPrevote
- cs.setProposal = cs.defaultSetProposal
-
- cs.updateToState(state)
- // Don't call scheduleRound0 yet.
- // We do that upon Start().
- cs.reconstructLastCommit(state)
- cs.BaseService = *cmn.NewBaseService(nil, "ConsensusState", cs)
- return cs
- }
-
- //----------------------------------------
- // Public interface
-
- // SetLogger implements Service.
- func (cs *ConsensusState) SetLogger(l log.Logger) {
- cs.BaseService.Logger = l
- cs.timeoutTicker.SetLogger(l)
- }
-
- // SetEventSwitch implements events.Eventable
- func (cs *ConsensusState) SetEventSwitch(evsw types.EventSwitch) {
- cs.evsw = evsw
- }
-
- // String returns a string.
- func (cs *ConsensusState) String() string {
- // better not to access shared variables
- return cmn.Fmt("ConsensusState") //(H:%v R:%v S:%v", cs.Height, cs.Round, cs.Step)
- }
-
- // GetState returns a copy of the chain state.
- func (cs *ConsensusState) GetState() *sm.State {
- cs.mtx.Lock()
- defer cs.mtx.Unlock()
- return cs.state.Copy()
- }
-
- // GetRoundState returns a copy of the internal consensus state.
- func (cs *ConsensusState) GetRoundState() *RoundState {
- cs.mtx.Lock()
- defer cs.mtx.Unlock()
- return cs.getRoundState()
- }
-
- func (cs *ConsensusState) getRoundState() *RoundState {
- rs := cs.RoundState // copy
- return &rs
- }
-
- // GetValidators returns a copy of the current validators.
- func (cs *ConsensusState) GetValidators() (int, []*types.Validator) {
- cs.mtx.Lock()
- defer cs.mtx.Unlock()
- return cs.state.LastBlockHeight, cs.state.Validators.Copy().Validators
- }
-
- // SetPrivValidator sets the private validator account for signing votes.
- func (cs *ConsensusState) SetPrivValidator(priv PrivValidator) {
- cs.mtx.Lock()
- defer cs.mtx.Unlock()
- cs.privValidator = priv
- }
-
- // SetTimeoutTicker sets the local timer. It may be useful to overwrite for testing.
- func (cs *ConsensusState) SetTimeoutTicker(timeoutTicker TimeoutTicker) {
- cs.mtx.Lock()
- defer cs.mtx.Unlock()
- cs.timeoutTicker = timeoutTicker
- }
-
- // LoadCommit loads the commit for a given height.
- func (cs *ConsensusState) LoadCommit(height int) *types.Commit {
- cs.mtx.Lock()
- defer cs.mtx.Unlock()
- if height == cs.blockStore.Height() {
- return cs.blockStore.LoadSeenCommit(height)
- }
- return cs.blockStore.LoadBlockCommit(height)
- }
-
- // OnStart implements cmn.Service.
- // It loads the latest state via the WAL, and starts the timeout and receive routines.
- func (cs *ConsensusState) OnStart() error {
-
- walFile := cs.config.WalFile()
- if err := cs.OpenWAL(walFile); err != nil {
- cs.Logger.Error("Error loading ConsensusState wal", "err", err.Error())
- return err
- }
-
- // we need the timeoutRoutine for replay so
- // we don't block on the tick chan.
- // NOTE: we will get a build up of garbage go routines
- // firing on the tockChan until the receiveRoutine is started
- // to deal with them (by that point, at most one will be valid)
- cs.timeoutTicker.Start()
-
- // we may have lost some votes if the process crashed
- // reload from consensus log to catchup
- if err := cs.catchupReplay(cs.Height); err != nil {
- cs.Logger.Error("Error on catchup replay. Proceeding to start ConsensusState anyway", "err", err.Error())
- // NOTE: if we ever do return an error here,
- // make sure to stop the timeoutTicker
- }
-
- // now start the receiveRoutine
- go cs.receiveRoutine(0)
-
- // schedule the first round!
- // use GetRoundState so we don't race the receiveRoutine for access
- cs.scheduleRound0(cs.GetRoundState())
-
- return nil
- }
-
- // timeoutRoutine: receive requests for timeouts on tickChan and fire timeouts on tockChan
- // receiveRoutine: serializes processing of proposoals, block parts, votes; coordinates state transitions
- func (cs *ConsensusState) startRoutines(maxSteps int) {
- cs.timeoutTicker.Start()
- go cs.receiveRoutine(maxSteps)
- }
-
- // OnStop implements cmn.Service. It stops all routines and waits for the WAL to finish.
- func (cs *ConsensusState) OnStop() {
- cs.BaseService.OnStop()
-
- cs.timeoutTicker.Stop()
-
- // Make BaseService.Wait() wait until cs.wal.Wait()
- if cs.wal != nil && cs.IsRunning() {
- cs.wal.Wait()
- }
- }
-
- // Wait waits for the the main routine to return.
- // NOTE: be sure to Stop() the event switch and drain
- // any event channels or this may deadlock
- func (cs *ConsensusState) Wait() {
- <-cs.done
- }
-
- // OpenWAL opens a file to log all consensus messages and timeouts for deterministic accountability
- func (cs *ConsensusState) OpenWAL(walFile string) (err error) {
- err = cmn.EnsureDir(path.Dir(walFile), 0700)
- if err != nil {
- cs.Logger.Error("Error ensuring ConsensusState wal dir", "err", err.Error())
- return err
- }
-
- cs.mtx.Lock()
- defer cs.mtx.Unlock()
- wal, err := NewWAL(walFile, cs.config.WalLight)
- if err != nil {
- return err
- }
- wal.SetLogger(cs.Logger.With("wal", walFile))
- if _, err := wal.Start(); err != nil {
- return err
- }
- cs.wal = wal
- return nil
- }
-
- //------------------------------------------------------------
- // Public interface for passing messages into the consensus state, possibly causing a state transition.
- // If peerKey == "", the msg is considered internal.
- // Messages are added to the appropriate queue (peer or internal).
- // If the queue is full, the function may block.
- // TODO: should these return anything or let callers just use events?
-
- // AddVote inputs a vote.
- func (cs *ConsensusState) AddVote(vote *types.Vote, peerKey string) (added bool, err error) {
- if peerKey == "" {
- cs.internalMsgQueue <- msgInfo{&VoteMessage{vote}, ""}
- } else {
- cs.peerMsgQueue <- msgInfo{&VoteMessage{vote}, peerKey}
- }
-
- // TODO: wait for event?!
- return false, nil
- }
-
- // SetProposal inputs a proposal.
- func (cs *ConsensusState) SetProposal(proposal *types.Proposal, peerKey string) error {
-
- if peerKey == "" {
- cs.internalMsgQueue <- msgInfo{&ProposalMessage{proposal}, ""}
- } else {
- cs.peerMsgQueue <- msgInfo{&ProposalMessage{proposal}, peerKey}
- }
-
- // TODO: wait for event?!
- return nil
- }
-
- // AddProposalBlockPart inputs a part of the proposal block.
- func (cs *ConsensusState) AddProposalBlockPart(height, round int, part *types.Part, peerKey string) error {
-
- if peerKey == "" {
- cs.internalMsgQueue <- msgInfo{&BlockPartMessage{height, round, part}, ""}
- } else {
- cs.peerMsgQueue <- msgInfo{&BlockPartMessage{height, round, part}, peerKey}
- }
-
- // TODO: wait for event?!
- return nil
- }
-
- // SetProposalAndBlock inputs the proposal and all block parts.
- func (cs *ConsensusState) SetProposalAndBlock(proposal *types.Proposal, block *types.Block, parts *types.PartSet, peerKey string) error {
- cs.SetProposal(proposal, peerKey)
- for i := 0; i < parts.Total(); i++ {
- part := parts.GetPart(i)
- cs.AddProposalBlockPart(proposal.Height, proposal.Round, part, peerKey)
- }
- return nil // TODO errors
- }
-
- //------------------------------------------------------------
- // internal functions for managing the state
-
- func (cs *ConsensusState) updateHeight(height int) {
- cs.Height = height
- }
-
- func (cs *ConsensusState) updateRoundStep(round int, step RoundStepType) {
- cs.Round = round
- cs.Step = step
- }
-
- // enterNewRound(height, 0) at cs.StartTime.
- func (cs *ConsensusState) scheduleRound0(rs *RoundState) {
- //cs.Logger.Info("scheduleRound0", "now", time.Now(), "startTime", cs.StartTime)
- sleepDuration := rs.StartTime.Sub(time.Now())
- cs.scheduleTimeout(sleepDuration, rs.Height, 0, RoundStepNewHeight)
- }
-
- // Attempt to schedule a timeout (by sending timeoutInfo on the tickChan)
- func (cs *ConsensusState) scheduleTimeout(duration time.Duration, height, round int, step RoundStepType) {
- cs.timeoutTicker.ScheduleTimeout(timeoutInfo{duration, height, round, step})
- }
-
- // send a msg into the receiveRoutine regarding our own proposal, block part, or vote
- func (cs *ConsensusState) sendInternalMessage(mi msgInfo) {
- select {
- case cs.internalMsgQueue <- mi:
- default:
- // NOTE: using the go-routine means our votes can
- // be processed out of order.
- // TODO: use CList here for strict determinism and
- // attempt push to internalMsgQueue in receiveRoutine
- cs.Logger.Info("Internal msg queue is full. Using a go-routine")
- go func() { cs.internalMsgQueue <- mi }()
- }
- }
-
- // Reconstruct LastCommit from SeenCommit, which we saved along with the block,
- // (which happens even before saving the state)
- func (cs *ConsensusState) reconstructLastCommit(state *sm.State) {
- if state.LastBlockHeight == 0 {
- return
- }
- seenCommit := cs.blockStore.LoadSeenCommit(state.LastBlockHeight)
- lastPrecommits := types.NewVoteSet(cs.state.ChainID, state.LastBlockHeight, seenCommit.Round(), types.VoteTypePrecommit, state.LastValidators)
- for _, precommit := range seenCommit.Precommits {
- if precommit == nil {
- continue
- }
- added, err := lastPrecommits.AddVote(precommit)
- if !added || err != nil {
- cmn.PanicCrisis(cmn.Fmt("Failed to reconstruct LastCommit: %v", err))
- }
- }
- if !lastPrecommits.HasTwoThirdsMajority() {
- cmn.PanicSanity("Failed to reconstruct LastCommit: Does not have +2/3 maj")
- }
- cs.LastCommit = lastPrecommits
- }
-
- // Updates ConsensusState and increments height to match that of state.
- // The round becomes 0 and cs.Step becomes RoundStepNewHeight.
- func (cs *ConsensusState) updateToState(state *sm.State) {
- if cs.CommitRound > -1 && 0 < cs.Height && cs.Height != state.LastBlockHeight {
- cmn.PanicSanity(cmn.Fmt("updateToState() expected state height of %v but found %v",
- cs.Height, state.LastBlockHeight))
- }
- if cs.state != nil && cs.state.LastBlockHeight+1 != cs.Height {
- // This might happen when someone else is mutating cs.state.
- // Someone forgot to pass in state.Copy() somewhere?!
- cmn.PanicSanity(cmn.Fmt("Inconsistent cs.state.LastBlockHeight+1 %v vs cs.Height %v",
- cs.state.LastBlockHeight+1, cs.Height))
- }
-
- // If state isn't further out than cs.state, just ignore.
- // This happens when SwitchToConsensus() is called in the reactor.
- // We don't want to reset e.g. the Votes.
- if cs.state != nil && (state.LastBlockHeight <= cs.state.LastBlockHeight) {
- cs.Logger.Info("Ignoring updateToState()", "newHeight", state.LastBlockHeight+1, "oldHeight", cs.state.LastBlockHeight+1)
- return
- }
-
- // Reset fields based on state.
- validators := state.Validators
- lastPrecommits := (*types.VoteSet)(nil)
- if cs.CommitRound > -1 && cs.Votes != nil {
- if !cs.Votes.Precommits(cs.CommitRound).HasTwoThirdsMajority() {
- cmn.PanicSanity("updateToState(state) called but last Precommit round didn't have +2/3")
- }
- lastPrecommits = cs.Votes.Precommits(cs.CommitRound)
- }
-
- // Next desired block height
- height := state.LastBlockHeight + 1
-
- // RoundState fields
- cs.updateHeight(height)
- cs.updateRoundStep(0, RoundStepNewHeight)
- if cs.CommitTime.IsZero() {
- // "Now" makes it easier to sync up dev nodes.
- // We add timeoutCommit to allow transactions
- // to be gathered for the first block.
- // And alternative solution that relies on clocks:
- // cs.StartTime = state.LastBlockTime.Add(timeoutCommit)
- cs.StartTime = cs.config.Commit(time.Now())
- } else {
- cs.StartTime = cs.config.Commit(cs.CommitTime)
- }
- cs.Validators = validators
- cs.Proposal = nil
- cs.ProposalBlock = nil
- cs.ProposalBlockParts = nil
- cs.LockedRound = 0
- cs.LockedBlock = nil
- cs.LockedBlockParts = nil
- cs.Votes = NewHeightVoteSet(state.ChainID, height, validators)
- cs.CommitRound = -1
- cs.LastCommit = lastPrecommits
- cs.LastValidators = state.LastValidators
-
- cs.state = state
-
- // Finally, broadcast RoundState
- cs.newStep()
- }
-
- func (cs *ConsensusState) newStep() {
- rs := cs.RoundStateEvent()
- cs.wal.Save(rs)
- cs.nSteps += 1
- // newStep is called by updateToStep in NewConsensusState before the evsw is set!
- if cs.evsw != nil {
- types.FireEventNewRoundStep(cs.evsw, rs)
- }
- }
-
- //-----------------------------------------
- // the main go routines
-
- // receiveRoutine handles messages which may cause state transitions.
- // it's argument (n) is the number of messages to process before exiting - use 0 to run forever
- // It keeps the RoundState and is the only thing that updates it.
- // Updates (state transitions) happen on timeouts, complete proposals, and 2/3 majorities
- func (cs *ConsensusState) receiveRoutine(maxSteps int) {
- for {
- if maxSteps > 0 {
- if cs.nSteps >= maxSteps {
- cs.Logger.Info("reached max steps. exiting receive routine")
- cs.nSteps = 0
- return
- }
- }
- rs := cs.RoundState
- var mi msgInfo
-
- select {
- case mi = <-cs.peerMsgQueue:
- cs.wal.Save(mi)
- // handles proposals, block parts, votes
- // may generate internal events (votes, complete proposals, 2/3 majorities)
- cs.handleMsg(mi, rs)
- case mi = <-cs.internalMsgQueue:
- cs.wal.Save(mi)
- // handles proposals, block parts, votes
- cs.handleMsg(mi, rs)
- case ti := <-cs.timeoutTicker.Chan(): // tockChan:
- cs.wal.Save(ti)
- // if the timeout is relevant to the rs
- // go to the next step
- cs.handleTimeout(ti, rs)
- case <-cs.Quit:
-
- // NOTE: the internalMsgQueue may have signed messages from our
- // priv_val that haven't hit the WAL, but its ok because
- // priv_val tracks LastSig
-
- // close wal now that we're done writing to it
- if cs.wal != nil {
- cs.wal.Stop()
- }
-
- close(cs.done)
- return
- }
- }
- }
-
- // state transitions on complete-proposal, 2/3-any, 2/3-one
- func (cs *ConsensusState) handleMsg(mi msgInfo, rs RoundState) {
- cs.mtx.Lock()
- defer cs.mtx.Unlock()
-
- var err error
- msg, peerKey := mi.Msg, mi.PeerKey
- switch msg := msg.(type) {
- case *ProposalMessage:
- // will not cause transition.
- // once proposal is set, we can receive block parts
- err = cs.setProposal(msg.Proposal)
- case *BlockPartMessage:
- // if the proposal is complete, we'll enterPrevote or tryFinalizeCommit
- _, err = cs.addProposalBlockPart(msg.Height, msg.Part, peerKey != "")
- if err != nil && msg.Round != cs.Round {
- err = nil
- }
- case *VoteMessage:
- // attempt to add the vote and dupeout the validator if its a duplicate signature
- // if the vote gives us a 2/3-any or 2/3-one, we transition
- err := cs.tryAddVote(msg.Vote, peerKey)
- if err == ErrAddingVote {
- // TODO: punish peer
- }
-
- // NOTE: the vote is broadcast to peers by the reactor listening
- // for vote events
-
- // TODO: If rs.Height == vote.Height && rs.Round < vote.Round,
- // the peer is sending us CatchupCommit precommits.
- // We could make note of this and help filter in broadcastHasVoteMessage().
- default:
- cs.Logger.Error("Unknown msg type", reflect.TypeOf(msg))
- }
- if err != nil {
- cs.Logger.Error("Error with msg", "type", reflect.TypeOf(msg), "peer", peerKey, "err", err, "msg", msg)
- }
- }
-
- func (cs *ConsensusState) handleTimeout(ti timeoutInfo, rs RoundState) {
- cs.Logger.Debug("Received tock", "timeout", ti.Duration, "height", ti.Height, "round", ti.Round, "step", ti.Step)
-
- // timeouts must be for current height, round, step
- if ti.Height != rs.Height || ti.Round < rs.Round || (ti.Round == rs.Round && ti.Step < rs.Step) {
- cs.Logger.Debug("Ignoring tock because we're ahead", "height", rs.Height, "round", rs.Round, "step", rs.Step)
- return
- }
-
- // the timeout will now cause a state transition
- cs.mtx.Lock()
- defer cs.mtx.Unlock()
-
- switch ti.Step {
- case RoundStepNewHeight:
- // NewRound event fired from enterNewRound.
- // XXX: should we fire timeout here (for timeout commit)?
- cs.enterNewRound(ti.Height, 0)
- case RoundStepPropose:
- types.FireEventTimeoutPropose(cs.evsw, cs.RoundStateEvent())
- cs.enterPrevote(ti.Height, ti.Round)
- case RoundStepPrevoteWait:
- types.FireEventTimeoutWait(cs.evsw, cs.RoundStateEvent())
- cs.enterPrecommit(ti.Height, ti.Round)
- case RoundStepPrecommitWait:
- types.FireEventTimeoutWait(cs.evsw, cs.RoundStateEvent())
- cs.enterNewRound(ti.Height, ti.Round+1)
- default:
- panic(cmn.Fmt("Invalid timeout step: %v", ti.Step))
- }
-
- }
-
- //-----------------------------------------------------------------------------
- // State functions
- // Used internally by handleTimeout and handleMsg to make state transitions
-
- // Enter: `timeoutNewHeight` by startTime (commitTime+timeoutCommit),
- // or, if SkipTimeout==true, after receiving all precommits from (height,round-1)
- // Enter: `timeoutPrecommits` after any +2/3 precommits from (height,round-1)
- // Enter: +2/3 precommits for nil at (height,round-1)
- // Enter: +2/3 prevotes any or +2/3 precommits for block or any from (height, round)
- // NOTE: cs.StartTime was already set for height.
- func (cs *ConsensusState) enterNewRound(height int, round int) {
- if cs.Height != height || round < cs.Round || (cs.Round == round && cs.Step != RoundStepNewHeight) {
- cs.Logger.Debug(cmn.Fmt("enterNewRound(%v/%v): Invalid args. Current step: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step))
- return
- }
-
- if now := time.Now(); cs.StartTime.After(now) {
- cs.Logger.Info("Need to set a buffer and log message here for sanity.", "startTime", cs.StartTime, "now", now)
- }
-
- cs.Logger.Info(cmn.Fmt("enterNewRound(%v/%v). Current: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step))
-
- // Increment validators if necessary
- validators := cs.Validators
- if cs.Round < round {
- validators = validators.Copy()
- validators.IncrementAccum(round - cs.Round)
- }
-
- // Setup new round
- // we don't fire newStep for this step,
- // but we fire an event, so update the round step first
- cs.updateRoundStep(round, RoundStepNewRound)
- cs.Validators = validators
- if round == 0 {
- // We've already reset these upon new height,
- // and meanwhile we might have received a proposal
- // for round 0.
- } else {
- cs.Proposal = nil
- cs.ProposalBlock = nil
- cs.ProposalBlockParts = nil
- }
- cs.Votes.SetRound(round + 1) // also track next round (round+1) to allow round-skipping
-
- types.FireEventNewRound(cs.evsw, cs.RoundStateEvent())
-
- // Immediately go to enterPropose.
- cs.enterPropose(height, round)
- }
-
- // Enter: from enterNewRound(height,round).
- func (cs *ConsensusState) enterPropose(height int, round int) {
- if cs.Height != height || round < cs.Round || (cs.Round == round && RoundStepPropose <= cs.Step) {
- cs.Logger.Debug(cmn.Fmt("enterPropose(%v/%v): Invalid args. Current step: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step))
- return
- }
- cs.Logger.Info(cmn.Fmt("enterPropose(%v/%v). Current: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step))
-
- defer func() {
- // Done enterPropose:
- cs.updateRoundStep(round, RoundStepPropose)
- cs.newStep()
-
- // If we have the whole proposal + POL, then goto Prevote now.
- // else, we'll enterPrevote when the rest of the proposal is received (in AddProposalBlockPart),
- // or else after timeoutPropose
- if cs.isProposalComplete() {
- cs.enterPrevote(height, cs.Round)
- }
- }()
-
- // If we don't get the proposal and all block parts quick enough, enterPrevote
- cs.scheduleTimeout(cs.config.Propose(round), height, round, RoundStepPropose)
-
- // Nothing more to do if we're not a validator
- if cs.privValidator == nil {
- cs.Logger.Debug("This node is not a validator")
- return
- }
-
- if !bytes.Equal(cs.Validators.GetProposer().Address, cs.privValidator.GetAddress()) {
- cs.Logger.Info("enterPropose: Not our turn to propose", "proposer", cs.Validators.GetProposer().Address, "privValidator", cs.privValidator)
- if cs.Validators.HasAddress(cs.privValidator.GetAddress()) {
- cs.Logger.Debug("This node is a validator")
- } else {
- cs.Logger.Debug("This node is not a validator")
- }
- } else {
- cs.Logger.Info("enterPropose: Our turn to propose", "proposer", cs.Validators.GetProposer().Address, "privValidator", cs.privValidator)
- cs.Logger.Debug("This node is a validator")
- cs.decideProposal(height, round)
- }
- }
-
- func (cs *ConsensusState) defaultDecideProposal(height, round int) {
- var block *types.Block
- var blockParts *types.PartSet
-
- // Decide on block
- if cs.LockedBlock != nil {
- // If we're locked onto a block, just choose that.
- block, blockParts = cs.LockedBlock, cs.LockedBlockParts
- } else {
- // Create a new proposal block from state/txs from the mempool.
- block, blockParts = cs.createProposalBlock()
- if block == nil { // on error
- return
- }
- }
-
- // Make proposal
- polRound, polBlockID := cs.Votes.POLInfo()
- proposal := types.NewProposal(height, round, blockParts.Header(), polRound, polBlockID)
- err := cs.privValidator.SignProposal(cs.state.ChainID, proposal)
- if err == nil {
- // Set fields
- /* fields set by setProposal and addBlockPart
- cs.Proposal = proposal
- cs.ProposalBlock = block
- cs.ProposalBlockParts = blockParts
- */
-
- // send proposal and block parts on internal msg queue
- cs.sendInternalMessage(msgInfo{&ProposalMessage{proposal}, ""})
- for i := 0; i < blockParts.Total(); i++ {
- part := blockParts.GetPart(i)
- cs.sendInternalMessage(msgInfo{&BlockPartMessage{cs.Height, cs.Round, part}, ""})
- }
- cs.Logger.Info("Signed proposal", "height", height, "round", round, "proposal", proposal)
- cs.Logger.Debug(cmn.Fmt("Signed proposal block: %v", block))
- } else {
- if !cs.replayMode {
- cs.Logger.Error("enterPropose: Error signing proposal", "height", height, "round", round, "err", err)
- }
- }
- }
-
- // Returns true if the proposal block is complete &&
- // (if POLRound was proposed, we have +2/3 prevotes from there).
- func (cs *ConsensusState) isProposalComplete() bool {
- if cs.Proposal == nil || cs.ProposalBlock == nil {
- return false
- }
- // we have the proposal. if there's a POLRound,
- // make sure we have the prevotes from it too
- if cs.Proposal.POLRound < 0 {
- return true
- } else {
- // if this is false the proposer is lying or we haven't received the POL yet
- return cs.Votes.Prevotes(cs.Proposal.POLRound).HasTwoThirdsMajority()
- }
- }
-
- // Create the next block to propose and return it.
- // Returns nil block upon error.
- // NOTE: keep it side-effect free for clarity.
- func (cs *ConsensusState) createProposalBlock() (block *types.Block, blockParts *types.PartSet) {
- var commit *types.Commit
- if cs.Height == 1 {
- // We're creating a proposal for the first block.
- // The commit is empty, but not nil.
- commit = &types.Commit{}
- } else if cs.LastCommit.HasTwoThirdsMajority() {
- // Make the commit from LastCommit
- commit = cs.LastCommit.MakeCommit()
- } else {
- // This shouldn't happen.
- cs.Logger.Error("enterPropose: Cannot propose anything: No commit for the previous block.")
- return
- }
-
- // Mempool validated transactions
- txs := cs.mempool.Reap(cs.config.MaxBlockSizeTxs)
-
- return types.MakeBlock(cs.Height, cs.state.ChainID, txs, commit,
- cs.state.LastBlockID, cs.state.Validators.Hash(), cs.state.AppHash, cs.config.BlockPartSize)
- }
-
- // Enter: `timeoutPropose` after entering Propose.
- // Enter: proposal block and POL is ready.
- // Enter: any +2/3 prevotes for future round.
- // Prevote for LockedBlock if we're locked, or ProposalBlock if valid.
- // Otherwise vote nil.
- func (cs *ConsensusState) enterPrevote(height int, round int) {
- if cs.Height != height || round < cs.Round || (cs.Round == round && RoundStepPrevote <= cs.Step) {
- cs.Logger.Debug(cmn.Fmt("enterPrevote(%v/%v): Invalid args. Current step: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step))
- return
- }
-
- defer func() {
- // Done enterPrevote:
- cs.updateRoundStep(round, RoundStepPrevote)
- cs.newStep()
- }()
-
- // fire event for how we got here
- if cs.isProposalComplete() {
- types.FireEventCompleteProposal(cs.evsw, cs.RoundStateEvent())
- } else {
- // we received +2/3 prevotes for a future round
- // TODO: catchup event?
- }
-
- cs.Logger.Info(cmn.Fmt("enterPrevote(%v/%v). Current: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step))
-
- // Sign and broadcast vote as necessary
- cs.doPrevote(height, round)
-
- // Once `addVote` hits any +2/3 prevotes, we will go to PrevoteWait
- // (so we have more time to try and collect +2/3 prevotes for a single block)
- }
-
- func (cs *ConsensusState) defaultDoPrevote(height int, round int) {
- logger := cs.Logger.With("height", height, "round", round)
- // If a block is locked, prevote that.
- if cs.LockedBlock != nil {
- logger.Info("enterPrevote: Block was locked")
- cs.signAddVote(types.VoteTypePrevote, cs.LockedBlock.Hash(), cs.LockedBlockParts.Header())
- return
- }
-
- // If ProposalBlock is nil, prevote nil.
- if cs.ProposalBlock == nil {
- logger.Info("enterPrevote: ProposalBlock is nil")
- cs.signAddVote(types.VoteTypePrevote, nil, types.PartSetHeader{})
- return
- }
-
- // Validate proposal block
- err := cs.state.ValidateBlock(cs.ProposalBlock)
- if err != nil {
- // ProposalBlock is invalid, prevote nil.
- logger.Error("enterPrevote: ProposalBlock is invalid", "err", err)
- cs.signAddVote(types.VoteTypePrevote, nil, types.PartSetHeader{})
- return
- }
-
- // Prevote cs.ProposalBlock
- // NOTE: the proposal signature is validated when it is received,
- // and the proposal block parts are validated as they are received (against the merkle hash in the proposal)
- logger.Info("enterPrevote: ProposalBlock is valid")
- cs.signAddVote(types.VoteTypePrevote, cs.ProposalBlock.Hash(), cs.ProposalBlockParts.Header())
- }
-
- // Enter: any +2/3 prevotes at next round.
- func (cs *ConsensusState) enterPrevoteWait(height int, round int) {
- if cs.Height != height || round < cs.Round || (cs.Round == round && RoundStepPrevoteWait <= cs.Step) {
- cs.Logger.Debug(cmn.Fmt("enterPrevoteWait(%v/%v): Invalid args. Current step: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step))
- return
- }
- if !cs.Votes.Prevotes(round).HasTwoThirdsAny() {
- cmn.PanicSanity(cmn.Fmt("enterPrevoteWait(%v/%v), but Prevotes does not have any +2/3 votes", height, round))
- }
- cs.Logger.Info(cmn.Fmt("enterPrevoteWait(%v/%v). Current: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step))
-
- defer func() {
- // Done enterPrevoteWait:
- cs.updateRoundStep(round, RoundStepPrevoteWait)
- cs.newStep()
- }()
-
- // Wait for some more prevotes; enterPrecommit
- cs.scheduleTimeout(cs.config.Prevote(round), height, round, RoundStepPrevoteWait)
- }
-
- // Enter: `timeoutPrevote` after any +2/3 prevotes.
- // Enter: +2/3 precomits for block or nil.
- // Enter: any +2/3 precommits for next round.
- // Lock & precommit the ProposalBlock if we have enough prevotes for it (a POL in this round)
- // else, unlock an existing lock and precommit nil if +2/3 of prevotes were nil,
- // else, precommit nil otherwise.
- func (cs *ConsensusState) enterPrecommit(height int, round int) {
- if cs.Height != height || round < cs.Round || (cs.Round == round && RoundStepPrecommit <= cs.Step) {
- cs.Logger.Debug(cmn.Fmt("enterPrecommit(%v/%v): Invalid args. Current step: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step))
- return
- }
-
- cs.Logger.Info(cmn.Fmt("enterPrecommit(%v/%v). Current: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step))
-
- defer func() {
- // Done enterPrecommit:
- cs.updateRoundStep(round, RoundStepPrecommit)
- cs.newStep()
- }()
-
- blockID, ok := cs.Votes.Prevotes(round).TwoThirdsMajority()
-
- // If we don't have a polka, we must precommit nil
- if !ok {
- if cs.LockedBlock != nil {
- cs.Logger.Info("enterPrecommit: No +2/3 prevotes during enterPrecommit while we're locked. Precommitting nil")
- } else {
- cs.Logger.Info("enterPrecommit: No +2/3 prevotes during enterPrecommit. Precommitting nil.")
- }
- cs.signAddVote(types.VoteTypePrecommit, nil, types.PartSetHeader{})
- return
- }
-
- // At this point +2/3 prevoted for a particular block or nil
- types.FireEventPolka(cs.evsw, cs.RoundStateEvent())
-
- // the latest POLRound should be this round
- polRound, _ := cs.Votes.POLInfo()
- if polRound < round {
- cmn.PanicSanity(cmn.Fmt("This POLRound should be %v but got %", round, polRound))
- }
-
- // +2/3 prevoted nil. Unlock and precommit nil.
- if len(blockID.Hash) == 0 {
- if cs.LockedBlock == nil {
- cs.Logger.Info("enterPrecommit: +2/3 prevoted for nil.")
- } else {
- cs.Logger.Info("enterPrecommit: +2/3 prevoted for nil. Unlocking")
- cs.LockedRound = 0
- cs.LockedBlock = nil
- cs.LockedBlockParts = nil
- types.FireEventUnlock(cs.evsw, cs.RoundStateEvent())
- }
- cs.signAddVote(types.VoteTypePrecommit, nil, types.PartSetHeader{})
- return
- }
-
- // At this point, +2/3 prevoted for a particular block.
-
- // If we're already locked on that block, precommit it, and update the LockedRound
- if cs.LockedBlock.HashesTo(blockID.Hash) {
- cs.Logger.Info("enterPrecommit: +2/3 prevoted locked block. Relocking")
- cs.LockedRound = round
- types.FireEventRelock(cs.evsw, cs.RoundStateEvent())
- cs.signAddVote(types.VoteTypePrecommit, blockID.Hash, blockID.PartsHeader)
- return
- }
-
- // If +2/3 prevoted for proposal block, stage and precommit it
- if cs.ProposalBlock.HashesTo(blockID.Hash) {
- cs.Logger.Info("enterPrecommit: +2/3 prevoted proposal block. Locking", "hash", blockID.Hash)
- // Validate the block.
- if err := cs.state.ValidateBlock(cs.ProposalBlock); err != nil {
- cmn.PanicConsensus(cmn.Fmt("enterPrecommit: +2/3 prevoted for an invalid block: %v", err))
- }
- cs.LockedRound = round
- cs.LockedBlock = cs.ProposalBlock
- cs.LockedBlockParts = cs.ProposalBlockParts
- types.FireEventLock(cs.evsw, cs.RoundStateEvent())
- cs.signAddVote(types.VoteTypePrecommit, blockID.Hash, blockID.PartsHeader)
- return
- }
-
- // There was a polka in this round for a block we don't have.
- // Fetch that block, unlock, and precommit nil.
- // The +2/3 prevotes for this round is the POL for our unlock.
- // TODO: In the future save the POL prevotes for justification.
- cs.LockedRound = 0
- cs.LockedBlock = nil
- cs.LockedBlockParts = nil
- if !cs.ProposalBlockParts.HasHeader(blockID.PartsHeader) {
- cs.ProposalBlock = nil
- cs.ProposalBlockParts = types.NewPartSetFromHeader(blockID.PartsHeader)
- }
- types.FireEventUnlock(cs.evsw, cs.RoundStateEvent())
- cs.signAddVote(types.VoteTypePrecommit, nil, types.PartSetHeader{})
- }
-
- // Enter: any +2/3 precommits for next round.
- func (cs *ConsensusState) enterPrecommitWait(height int, round int) {
- if cs.Height != height || round < cs.Round || (cs.Round == round && RoundStepPrecommitWait <= cs.Step) {
- cs.Logger.Debug(cmn.Fmt("enterPrecommitWait(%v/%v): Invalid args. Current step: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step))
- return
- }
- if !cs.Votes.Precommits(round).HasTwoThirdsAny() {
- cmn.PanicSanity(cmn.Fmt("enterPrecommitWait(%v/%v), but Precommits does not have any +2/3 votes", height, round))
- }
- cs.Logger.Info(cmn.Fmt("enterPrecommitWait(%v/%v). Current: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step))
-
- defer func() {
- // Done enterPrecommitWait:
- cs.updateRoundStep(round, RoundStepPrecommitWait)
- cs.newStep()
- }()
-
- // Wait for some more precommits; enterNewRound
- cs.scheduleTimeout(cs.config.Precommit(round), height, round, RoundStepPrecommitWait)
-
- }
-
- // Enter: +2/3 precommits for block
- func (cs *ConsensusState) enterCommit(height int, commitRound int) {
- if cs.Height != height || RoundStepCommit <= cs.Step {
- cs.Logger.Debug(cmn.Fmt("enterCommit(%v/%v): Invalid args. Current step: %v/%v/%v", height, commitRound, cs.Height, cs.Round, cs.Step))
- return
- }
- cs.Logger.Info(cmn.Fmt("enterCommit(%v/%v). Current: %v/%v/%v", height, commitRound, cs.Height, cs.Round, cs.Step))
-
- defer func() {
- // Done enterCommit:
- // keep cs.Round the same, commitRound points to the right Precommits set.
- cs.updateRoundStep(cs.Round, RoundStepCommit)
- cs.CommitRound = commitRound
- cs.CommitTime = time.Now()
- cs.newStep()
-
- // Maybe finalize immediately.
- cs.tryFinalizeCommit(height)
- }()
-
- blockID, ok := cs.Votes.Precommits(commitRound).TwoThirdsMajority()
- if !ok {
- cmn.PanicSanity("RunActionCommit() expects +2/3 precommits")
- }
-
- // The Locked* fields no longer matter.
- // Move them over to ProposalBlock if they match the commit hash,
- // otherwise they'll be cleared in updateToState.
- if cs.LockedBlock.HashesTo(blockID.Hash) {
- cs.ProposalBlock = cs.LockedBlock
- cs.ProposalBlockParts = cs.LockedBlockParts
- }
-
- // If we don't have the block being committed, set up to get it.
- if !cs.ProposalBlock.HashesTo(blockID.Hash) {
- if !cs.ProposalBlockParts.HasHeader(blockID.PartsHeader) {
- // We're getting the wrong block.
- // Set up ProposalBlockParts and keep waiting.
- cs.ProposalBlock = nil
- cs.ProposalBlockParts = types.NewPartSetFromHeader(blockID.PartsHeader)
- } else {
- // We just need to keep waiting.
- }
- }
- }
-
- // If we have the block AND +2/3 commits for it, finalize.
- func (cs *ConsensusState) tryFinalizeCommit(height int) {
- if cs.Height != height {
- cmn.PanicSanity(cmn.Fmt("tryFinalizeCommit() cs.Height: %v vs height: %v", cs.Height, height))
- }
-
- blockID, ok := cs.Votes.Precommits(cs.CommitRound).TwoThirdsMajority()
- if !ok || len(blockID.Hash) == 0 {
- cs.Logger.Error("Attempt to finalize failed. There was no +2/3 majority, or +2/3 was for <nil>.", "height", height)
- return
- }
- if !cs.ProposalBlock.HashesTo(blockID.Hash) {
- // TODO: this happens every time if we're not a validator (ugly logs)
- // TODO: ^^ wait, why does it matter that we're a validator?
- cs.Logger.Error("Attempt to finalize failed. We don't have the commit block.", "height", height, "proposal-block", cs.ProposalBlock.Hash(), "commit-block", blockID.Hash)
- return
- }
-
- // go
- cs.finalizeCommit(height)
- }
-
- // Increment height and goto RoundStepNewHeight
- func (cs *ConsensusState) finalizeCommit(height int) {
- if cs.Height != height || cs.Step != RoundStepCommit {
- cs.Logger.Debug(cmn.Fmt("finalizeCommit(%v): Invalid args. Current step: %v/%v/%v", height, cs.Height, cs.Round, cs.Step))
- return
- }
-
- blockID, ok := cs.Votes.Precommits(cs.CommitRound).TwoThirdsMajority()
- block, blockParts := cs.ProposalBlock, cs.ProposalBlockParts
-
- if !ok {
- cmn.PanicSanity(cmn.Fmt("Cannot finalizeCommit, commit does not have two thirds majority"))
- }
- if !blockParts.HasHeader(blockID.PartsHeader) {
- cmn.PanicSanity(cmn.Fmt("Expected ProposalBlockParts header to be commit header"))
- }
- if !block.HashesTo(blockID.Hash) {
- cmn.PanicSanity(cmn.Fmt("Cannot finalizeCommit, ProposalBlock does not hash to commit hash"))
- }
- if err := cs.state.ValidateBlock(block); err != nil {
- cmn.PanicConsensus(cmn.Fmt("+2/3 committed an invalid block: %v", err))
- }
-
- cs.Logger.Info(cmn.Fmt("Finalizing commit of block with %d txs", block.NumTxs),
- "height", block.Height, "hash", block.Hash(), "root", block.AppHash)
- cs.Logger.Info(cmn.Fmt("%v", block))
-
- fail.Fail() // XXX
-
- // Save to blockStore.
- if cs.blockStore.Height() < block.Height {
- // NOTE: the seenCommit is local justification to commit this block,
- // but may differ from the LastCommit included in the next block
- precommits := cs.Votes.Precommits(cs.CommitRound)
- seenCommit := precommits.MakeCommit()
- cs.blockStore.SaveBlock(block, blockParts, seenCommit)
- } else {
- // Happens during replay if we already saved the block but didn't commit
- cs.Logger.Info("Calling finalizeCommit on already stored block", "height", block.Height)
- }
-
- fail.Fail() // XXX
-
- // Finish writing to the WAL for this height.
- // NOTE: If we fail before writing this, we'll never write it,
- // and just recover by running ApplyBlock in the Handshake.
- // If we moved it before persisting the block, we'd have to allow
- // WAL replay for blocks with an #ENDHEIGHT
- // As is, ConsensusState should not be started again
- // until we successfully call ApplyBlock (ie. here or in Handshake after restart)
- if cs.wal != nil {
- cs.wal.writeEndHeight(height)
- }
-
- fail.Fail() // XXX
-
- // Create a copy of the state for staging
- // and an event cache for txs
- stateCopy := cs.state.Copy()
- eventCache := types.NewEventCache(cs.evsw)
-
- // Execute and commit the block, update and save the state, and update the mempool.
- // All calls to the proxyAppConn come here.
- // NOTE: the block.AppHash wont reflect these txs until the next block
- err := stateCopy.ApplyBlock(eventCache, cs.proxyAppConn, block, blockParts.Header(), cs.mempool)
- if err != nil {
- cs.Logger.Error("Error on ApplyBlock. Did the application crash? Please restart tendermint", "err", err)
- return
- }
-
- fail.Fail() // XXX
-
- // Fire event for new block.
- // NOTE: If we fail before firing, these events will never fire
- //
- // TODO: Either
- // * Fire before persisting state, in ApplyBlock
- // * Fire on start up if we haven't written any new WAL msgs
- // Both options mean we may fire more than once. Is that fine ?
- types.FireEventNewBlock(cs.evsw, types.EventDataNewBlock{block})
- types.FireEventNewBlockHeader(cs.evsw, types.EventDataNewBlockHeader{block.Header})
- eventCache.Flush()
-
- fail.Fail() // XXX
-
- // NewHeightStep!
- cs.updateToState(stateCopy)
-
- fail.Fail() // XXX
-
- // cs.StartTime is already set.
- // Schedule Round0 to start soon.
- cs.scheduleRound0(&cs.RoundState)
-
- // By here,
- // * cs.Height has been increment to height+1
- // * cs.Step is now RoundStepNewHeight
- // * cs.StartTime is set to when we will start round0.
- }
-
- //-----------------------------------------------------------------------------
-
- func (cs *ConsensusState) defaultSetProposal(proposal *types.Proposal) error {
- // Already have one
- // TODO: possibly catch double proposals
- if cs.Proposal != nil {
- return nil
- }
-
- // Does not apply
- if proposal.Height != cs.Height || proposal.Round != cs.Round {
- return nil
- }
-
- // We don't care about the proposal if we're already in RoundStepCommit.
- if RoundStepCommit <= cs.Step {
- return nil
- }
-
- // Verify POLRound, which must be -1 or between 0 and proposal.Round exclusive.
- if proposal.POLRound != -1 &&
- (proposal.POLRound < 0 || proposal.Round <= proposal.POLRound) {
- return ErrInvalidProposalPOLRound
- }
-
- // Verify signature
- if !cs.Validators.GetProposer().PubKey.VerifyBytes(types.SignBytes(cs.state.ChainID, proposal), proposal.Signature) {
- return ErrInvalidProposalSignature
- }
-
- cs.Proposal = proposal
- cs.ProposalBlockParts = types.NewPartSetFromHeader(proposal.BlockPartsHeader)
- return nil
- }
-
- // NOTE: block is not necessarily valid.
- // Asynchronously triggers either enterPrevote (before we timeout of propose) or tryFinalizeCommit, once we have the full block.
- func (cs *ConsensusState) addProposalBlockPart(height int, part *types.Part, verify bool) (added bool, err error) {
- // Blocks might be reused, so round mismatch is OK
- if cs.Height != height {
- return false, nil
- }
-
- // We're not expecting a block part.
- if cs.ProposalBlockParts == nil {
- return false, nil // TODO: bad peer? Return error?
- }
-
- added, err = cs.ProposalBlockParts.AddPart(part, verify)
- if err != nil {
- return added, err
- }
- if added && cs.ProposalBlockParts.IsComplete() {
- // Added and completed!
- var n int
- var err error
- cs.ProposalBlock = wire.ReadBinary(&types.Block{}, cs.ProposalBlockParts.GetReader(), types.MaxBlockSize, &n, &err).(*types.Block)
- // NOTE: it's possible to receive complete proposal blocks for future rounds without having the proposal
- cs.Logger.Info("Received complete proposal block", "height", cs.ProposalBlock.Height, "hash", cs.ProposalBlock.Hash())
- if cs.Step == RoundStepPropose && cs.isProposalComplete() {
- // Move onto the next step
- cs.enterPrevote(height, cs.Round)
- } else if cs.Step == RoundStepCommit {
- // If we're waiting on the proposal block...
- cs.tryFinalizeCommit(height)
- }
- return true, err
- }
- return added, nil
- }
-
- // Attempt to add the vote. if its a duplicate signature, dupeout the validator
- func (cs *ConsensusState) tryAddVote(vote *types.Vote, peerKey string) error {
- _, err := cs.addVote(vote, peerKey)
- if err != nil {
- // If the vote height is off, we'll just ignore it,
- // But if it's a conflicting sig, broadcast evidence tx for slashing.
- // If it's otherwise invalid, punish peer.
- if err == ErrVoteHeightMismatch {
- return err
- } else if _, ok := err.(*types.ErrVoteConflictingVotes); ok {
- if bytes.Equal(vote.ValidatorAddress, cs.privValidator.GetAddress()) {
- cs.Logger.Error("Found conflicting vote from ourselves. Did you unsafe_reset a validator?", "height", vote.Height, "round", vote.Round, "type", vote.Type)
- return err
- }
- cs.Logger.Error("Found conflicting vote. Publish evidence (TODO)", "height", vote.Height, "round", vote.Round, "type", vote.Type, "valAddr", vote.ValidatorAddress, "valIndex", vote.ValidatorIndex)
-
- // TODO: track evidence for inclusion in a block
-
- return err
- } else {
- // Probably an invalid signature. Bad peer.
- cs.Logger.Error("Error attempting to add vote", "err", err)
- return ErrAddingVote
- }
- }
- return nil
- }
-
- //-----------------------------------------------------------------------------
-
- func (cs *ConsensusState) addVote(vote *types.Vote, peerKey string) (added bool, err error) {
- cs.Logger.Debug("addVote", "voteHeight", vote.Height, "voteType", vote.Type, "valIndex", vote.ValidatorIndex, "csHeight", cs.Height)
-
- // A precommit for the previous height?
- // These come in while we wait timeoutCommit
- if vote.Height+1 == cs.Height {
- if !(cs.Step == RoundStepNewHeight && vote.Type == types.VoteTypePrecommit) {
- // TODO: give the reason ..
- // fmt.Errorf("tryAddVote: Wrong height, not a LastCommit straggler commit.")
- return added, ErrVoteHeightMismatch
- }
- added, err = cs.LastCommit.AddVote(vote)
- if added {
- cs.Logger.Info(cmn.Fmt("Added to lastPrecommits: %v", cs.LastCommit.StringShort()))
- types.FireEventVote(cs.evsw, types.EventDataVote{vote})
-
- // if we can skip timeoutCommit and have all the votes now,
- if cs.config.SkipTimeoutCommit && cs.LastCommit.HasAll() {
- // go straight to new round (skip timeout commit)
- // cs.scheduleTimeout(time.Duration(0), cs.Height, 0, RoundStepNewHeight)
- cs.enterNewRound(cs.Height, 0)
- }
- }
-
- return
- }
-
- // A prevote/precommit for this height?
- if vote.Height == cs.Height {
- height := cs.Height
- added, err = cs.Votes.AddVote(vote, peerKey)
- if added {
- types.FireEventVote(cs.evsw, types.EventDataVote{vote})
-
- switch vote.Type {
- case types.VoteTypePrevote:
- prevotes := cs.Votes.Prevotes(vote.Round)
- cs.Logger.Info("Added to prevote", "vote", vote, "prevotes", prevotes.StringShort())
- // First, unlock if prevotes is a valid POL.
- // >> lockRound < POLRound <= unlockOrChangeLockRound (see spec)
- // NOTE: If (lockRound < POLRound) but !(POLRound <= unlockOrChangeLockRound),
- // we'll still enterNewRound(H,vote.R) and enterPrecommit(H,vote.R) to process it
- // there.
- if (cs.LockedBlock != nil) && (cs.LockedRound < vote.Round) && (vote.Round <= cs.Round) {
- blockID, ok := prevotes.TwoThirdsMajority()
- if ok && !cs.LockedBlock.HashesTo(blockID.Hash) {
- cs.Logger.Info("Unlocking because of POL.", "lockedRound", cs.LockedRound, "POLRound", vote.Round)
- cs.LockedRound = 0
- cs.LockedBlock = nil
- cs.LockedBlockParts = nil
- types.FireEventUnlock(cs.evsw, cs.RoundStateEvent())
- }
- }
- if cs.Round <= vote.Round && prevotes.HasTwoThirdsAny() {
- // Round-skip over to PrevoteWait or goto Precommit.
- cs.enterNewRound(height, vote.Round) // if the vote is ahead of us
- if prevotes.HasTwoThirdsMajority() {
- cs.enterPrecommit(height, vote.Round)
- } else {
- cs.enterPrevote(height, vote.Round) // if the vote is ahead of us
- cs.enterPrevoteWait(height, vote.Round)
- }
- } else if cs.Proposal != nil && 0 <= cs.Proposal.POLRound && cs.Proposal.POLRound == vote.Round {
- // If the proposal is now complete, enter prevote of cs.Round.
- if cs.isProposalComplete() {
- cs.enterPrevote(height, cs.Round)
- }
- }
- case types.VoteTypePrecommit:
- precommits := cs.Votes.Precommits(vote.Round)
- cs.Logger.Info("Added to precommit", "vote", vote, "precommits", precommits.StringShort())
- blockID, ok := precommits.TwoThirdsMajority()
- if ok {
- if len(blockID.Hash) == 0 {
- cs.enterNewRound(height, vote.Round+1)
- } else {
- cs.enterNewRound(height, vote.Round)
- cs.enterPrecommit(height, vote.Round)
- cs.enterCommit(height, vote.Round)
-
- if cs.config.SkipTimeoutCommit && precommits.HasAll() {
- // if we have all the votes now,
- // go straight to new round (skip timeout commit)
- // cs.scheduleTimeout(time.Duration(0), cs.Height, 0, RoundStepNewHeight)
- cs.enterNewRound(cs.Height, 0)
- }
-
- }
- } else if cs.Round <= vote.Round && precommits.HasTwoThirdsAny() {
- cs.enterNewRound(height, vote.Round)
- cs.enterPrecommit(height, vote.Round)
- cs.enterPrecommitWait(height, vote.Round)
- }
- default:
- cmn.PanicSanity(cmn.Fmt("Unexpected vote type %X", vote.Type)) // Should not happen.
- }
- }
- // Either duplicate, or error upon cs.Votes.AddByIndex()
- return
- } else {
- err = ErrVoteHeightMismatch
- }
-
- // Height mismatch, bad peer?
- cs.Logger.Info("Vote ignored and not added", "voteHeight", vote.Height, "csHeight", cs.Height, "err", err)
- return
- }
-
- func (cs *ConsensusState) signVote(type_ byte, hash []byte, header types.PartSetHeader) (*types.Vote, error) {
- addr := cs.privValidator.GetAddress()
- valIndex, _ := cs.Validators.GetByAddress(addr)
- vote := &types.Vote{
- ValidatorAddress: addr,
- ValidatorIndex: valIndex,
- Height: cs.Height,
- Round: cs.Round,
- Type: type_,
- BlockID: types.BlockID{hash, header},
- }
- err := cs.privValidator.SignVote(cs.state.ChainID, vote)
- return vote, err
- }
-
- // sign the vote and publish on internalMsgQueue
- func (cs *ConsensusState) signAddVote(type_ byte, hash []byte, header types.PartSetHeader) *types.Vote {
- // if we don't have a key or we're not in the validator set, do nothing
- if cs.privValidator == nil || !cs.Validators.HasAddress(cs.privValidator.GetAddress()) {
- return nil
- }
- vote, err := cs.signVote(type_, hash, header)
- if err == nil {
- cs.sendInternalMessage(msgInfo{&VoteMessage{vote}, ""})
- cs.Logger.Info("Signed and pushed vote", "height", cs.Height, "round", cs.Round, "vote", vote, "err", err)
- return vote
- } else {
- //if !cs.replayMode {
- cs.Logger.Error("Error signing vote", "height", cs.Height, "round", cs.Round, "vote", vote, "err", err)
- //}
- return nil
- }
- }
-
- //---------------------------------------------------------
-
- func CompareHRS(h1, r1 int, s1 RoundStepType, h2, r2 int, s2 RoundStepType) int {
- if h1 < h2 {
- return -1
- } else if h1 > h2 {
- return 1
- }
- if r1 < r2 {
- return -1
- } else if r1 > r2 {
- return 1
- }
- if s1 < s2 {
- return -1
- } else if s1 > s2 {
- return 1
- }
- return 0
- }
|