/* Consensus State Machine Overview: * Propose, Prevote, Precommit represent state machine stages. (aka RoundStep, or step). Each take a predetermined amount of time depending on the round number. * The Commit step can be entered by two means: 1. After the Precommit step, +2/3 Precommits were found 2. At any time, +2/3 Commits were found * Once in the Commit stage, two conditions must both be satisfied before proceeding to the next height NewHeight. * The Propose step of the next height does not begin until at least Delta duration *after* +2/3 Commits were found. The step stays at NewHeight until this timeout occurs before proceeding to Propose. +-------------------------------------+ | | v |(Wait til CommitTime + Delta) +-----------+ +-----+-----+ +----------> | Propose +--------------+ | NewHeight | | +-----------+ | +-----------+ | | ^ | | | | | | |(Else) v | +-----+-----+ +-----------+ | | Precommit | <------------------------+ Prevote | | +-----+-----+ +-----------+ | |(If +2/3 Precommits found) | | | | + (When +2/3 Commits found) | | | | v v | +------------------------------------------------------------------------------+ | Commit | | | | | | +----------------+ * Save Block | | | |Get Block Parts |---> * Stage Block +--+ + | | +----------------+ * Broadcast Commit | * Setup New Height | | | * Move Commits set to | | +--> LastCommits to continue | | | collecting commits | | +-----------------+ | * Broadcast New State | | |Get +2/3 Commits |--> * Set CommitTime +--+ | | +-----------------+ | | | +------------------------------------------------------------------------------+ */ package consensus import ( "bytes" "errors" "fmt" "math" "sync" "sync/atomic" "time" "github.com/tendermint/tendermint/account" "github.com/tendermint/tendermint/binary" bc "github.com/tendermint/tendermint/blockchain" . "github.com/tendermint/tendermint/common" "github.com/tendermint/tendermint/config" . "github.com/tendermint/tendermint/consensus/types" "github.com/tendermint/tendermint/events" mempl "github.com/tendermint/tendermint/mempool" sm "github.com/tendermint/tendermint/state" "github.com/tendermint/tendermint/types" ) const ( roundDeadlinePrevote = float64(1.0 / 3.0) // When the prevote is due. roundDeadlinePrecommit = float64(2.0 / 3.0) // When the precommit vote is due. ) var ( RoundDuration0 = 10 * time.Second // The first round is 60 seconds long. RoundDurationDelta = 3 * time.Second // Each successive round lasts 15 seconds longer. newHeightDelta = RoundDuration0 / 3 // The time to wait between commitTime and startTime of next consensus rounds. ) var ( ErrInvalidProposalSignature = errors.New("Error invalid proposal signature") ) //----------------------------------------------------------------------------- // RoundStepType enum type type RoundStepType uint8 // These must be numeric, ordered. const ( RoundStepNewHeight = RoundStepType(0x01) // Round0 for new height started, wait til CommitTime + Delta RoundStepNewRound = RoundStepType(0x02) // Pseudostep, immediately goes to RoundStepPropose RoundStepPropose = RoundStepType(0x03) // Did propose, gossip proposal RoundStepPrevote = RoundStepType(0x04) // Did prevote, gossip prevotes RoundStepPrecommit = RoundStepType(0x05) // Did precommit, gossip precommits RoundStepCommit = RoundStepType(0x06) // Entered commit state machine ) func (rs RoundStepType) String() string { switch rs { case RoundStepNewHeight: return "RoundStepNewHeight" case RoundStepNewRound: return "RoundStepNewRound" case RoundStepPropose: return "RoundStepPropose" case RoundStepPrevote: return "RoundStepPrevote" case RoundStepPrecommit: return "RoundStepPrecommit" case RoundStepCommit: return "RoundStepCommit" default: panic(Fmt("Unknown RoundStep %X", rs)) } } //----------------------------------------------------------------------------- // RoundAction enum type type RoundActionType string const ( RoundActionPropose = RoundActionType("PR") // Propose and goto RoundStepPropose RoundActionPrevote = RoundActionType("PV") // Prevote and goto RoundStepPrevote RoundActionPrecommit = RoundActionType("PC") // Precommit and goto RoundStepPrecommit RoundActionTryCommit = RoundActionType("TC") // Goto RoundStepCommit, or RoundStepPropose for next round. RoundActionCommit = RoundActionType("CM") // Goto RoundStepCommit upon +2/3 commits RoundActionTryFinalize = RoundActionType("TF") // Maybe goto RoundStepPropose for next round. ) func (rat RoundActionType) String() string { switch rat { case RoundActionPropose: return "RoundActionPropose" case RoundActionPrevote: return "RoundActionPrevote" case RoundActionPrecommit: return "RoundActionPrecommit" case RoundActionTryCommit: return "RoundActionTryCommit" case RoundActionCommit: return "RoundActionCommit" case RoundActionTryFinalize: return "RoundActionTryFinalize" default: panic(Fmt("Unknown RoundAction %X", rat)) } } //----------------------------------------------------------------------------- type RoundAction struct { Height uint // The block height for which consensus is reaching for. Round uint // The round number at given height. Action RoundActionType // Action to perform. } func (ra RoundAction) String() string { return Fmt("RoundAction{H:%v R:%v A:%v}", ra.Height, ra.Round, ra.Action) } //----------------------------------------------------------------------------- // Immutable when returned from ConsensusState.GetRoundState() type RoundState struct { Height uint // Height we are working on Round uint Step RoundStepType StartTime time.Time CommitTime time.Time // Time when +2/3 commits were found Validators *sm.ValidatorSet Proposal *Proposal ProposalBlock *types.Block ProposalBlockParts *types.PartSet ProposalPOL *POL ProposalPOLParts *types.PartSet LockedBlock *types.Block LockedBlockParts *types.PartSet LockedPOL *POL // Rarely needed, so no LockedPOLParts. Prevotes *VoteSet Precommits *VoteSet Commits *VoteSet LastCommits *VoteSet } func (rs *RoundState) String() string { return rs.StringIndented("") } 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 ProposalPOL: %v %v %s LockedBlock: %v %v %s LockedPOL: %v %s Prevotes: %v %s Precommits: %v %s Commits: %v %s LastCommits: %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.ProposalPOLParts.StringShort(), rs.ProposalPOL.StringShort(), indent, rs.LockedBlockParts.StringShort(), rs.LockedBlock.StringShort(), indent, rs.LockedPOL.StringShort(), indent, rs.Prevotes.StringIndented(indent+" "), indent, rs.Precommits.StringIndented(indent+" "), indent, rs.Commits.StringIndented(indent+" "), indent, rs.LastCommits.StringShort(), indent) } 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) } //----------------------------------------------------------------------------- // Tracks consensus state across block heights and rounds. type ConsensusState struct { started uint32 stopped uint32 quit chan struct{} blockStore *bc.BlockStore mempoolReactor *mempl.MempoolReactor privValidator *sm.PrivValidator runActionCh chan RoundAction newStepCh chan *RoundState mtx sync.Mutex RoundState state *sm.State // State until height-1. stagedBlock *types.Block // Cache last staged block. stagedState *sm.State // Cache result of staged block. lastCommitVoteHeight uint // Last called commitVoteBlock() or saveCommitVoteBlock() on. evsw events.Fireable evc *events.EventCache // set in stageBlock and passed into state } func NewConsensusState(state *sm.State, blockStore *bc.BlockStore, mempoolReactor *mempl.MempoolReactor) *ConsensusState { cs := &ConsensusState{ quit: make(chan struct{}), blockStore: blockStore, mempoolReactor: mempoolReactor, runActionCh: make(chan RoundAction, 1), newStepCh: make(chan *RoundState, 1), } cs.updateToState(state, true) return cs } func (cs *ConsensusState) GetState() *sm.State { cs.mtx.Lock() defer cs.mtx.Unlock() return cs.state.Copy() } 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 } func (cs *ConsensusState) NewStepCh() chan *RoundState { return cs.newStepCh } func (cs *ConsensusState) Start() { if atomic.CompareAndSwapUint32(&cs.started, 0, 1) { log.Info("Starting ConsensusState") go cs.stepTransitionRoutine() } } func (cs *ConsensusState) Stop() { if atomic.CompareAndSwapUint32(&cs.stopped, 0, 1) { log.Info("Stopping ConsensusState") close(cs.quit) } } func (cs *ConsensusState) IsStopped() bool { return atomic.LoadUint32(&cs.stopped) == 1 } func (cs *ConsensusState) queueAction(ra RoundAction) { go func() { cs.runActionCh <- ra }() } // Source of all round state transitions (and votes). func (cs *ConsensusState) stepTransitionRoutine() { // For clarity, all state transitions that happen after some timeout are here. // Schedule the next action by pushing a RoundAction{} to cs.runActionCh. scheduleNextAction := func() { // NOTE: rs must be fetched in the same callstack as the caller. rs := cs.getRoundState() go func() { // NOTE: We can push directly to runActionCh because // we're running in a separate goroutine, which avoids deadlocks. round, roundStartTime, RoundDuration, _, elapsedRatio := calcRoundInfo(rs.StartTime) log.Debug("Scheduling next action", "height", rs.Height, "round", round, "step", rs.Step, "roundStartTime", roundStartTime, "elapsedRatio", elapsedRatio) switch rs.Step { case RoundStepNewHeight: // We should run RoundActionPropose when rs.StartTime passes. if elapsedRatio < 0 { // startTime is in the future. time.Sleep(time.Duration((-1.0 * elapsedRatio) * float64(RoundDuration))) } cs.runActionCh <- RoundAction{rs.Height, rs.Round, RoundActionPropose} case RoundStepNewRound: // Pseudostep: Immediately goto propose. cs.runActionCh <- RoundAction{rs.Height, rs.Round, RoundActionPropose} case RoundStepPropose: // Wake up when it's time to vote. time.Sleep(time.Duration((roundDeadlinePrevote - elapsedRatio) * float64(RoundDuration))) cs.runActionCh <- RoundAction{rs.Height, rs.Round, RoundActionPrevote} case RoundStepPrevote: // Wake up when it's time to precommit. time.Sleep(time.Duration((roundDeadlinePrecommit - elapsedRatio) * float64(RoundDuration))) cs.runActionCh <- RoundAction{rs.Height, rs.Round, RoundActionPrecommit} case RoundStepPrecommit: // Wake up when the round is over. time.Sleep(time.Duration((1.0 - elapsedRatio) * float64(RoundDuration))) cs.runActionCh <- RoundAction{rs.Height, rs.Round, RoundActionTryCommit} case RoundStepCommit: // There's nothing to scheudle, we're waiting for // ProposalBlockParts.IsComplete() && // Commits.HasTwoThirdsMajority() panic("The next action from RoundStepCommit is not scheduled by time") default: panic("Should not happen") } }() } if cs.getRoundState().Step < RoundStepCommit { scheduleNextAction() } else { // Race condition with receipt of commits, maybe. // We shouldn't have to schedule anything. } // NOTE: All ConsensusState.RunAction*() calls come from here. // Since only one routine calls them, it is safe to assume that // the RoundState Height/Round/Step won't change concurrently. // However, other fields like Proposal could change concurrent // due to gossip routines. ACTION_LOOP: for { var roundAction RoundAction select { case roundAction = <-cs.runActionCh: case <-cs.quit: return } height, round, action := roundAction.Height, roundAction.Round, roundAction.Action rs := cs.GetRoundState() // Continue if action is not relevant if height != rs.Height { log.Debug("Discarding round action: Height mismatch", "height", rs.Height, "roundAction", roundAction) continue } // If action <= RoundActionPrecommit, the round must match too. if action <= RoundActionPrecommit && round != rs.Round { log.Debug("Discarding round action: Round mismatch", "round", rs.Round, "roundAction", roundAction) continue } log.Info("Running round action", "height", rs.Height, "round", rs.Round, "step", rs.Step, "roundAction", roundAction, "startTime", rs.StartTime) // Run action switch action { case RoundActionPropose: if rs.Step != RoundStepNewHeight && rs.Step != RoundStepNewRound { continue ACTION_LOOP } cs.RunActionPropose(rs.Height, rs.Round) scheduleNextAction() continue ACTION_LOOP case RoundActionPrevote: if rs.Step >= RoundStepPrevote { continue ACTION_LOOP } cs.RunActionPrevote(rs.Height, rs.Round) scheduleNextAction() continue ACTION_LOOP case RoundActionPrecommit: if rs.Step >= RoundStepPrecommit { continue ACTION_LOOP } cs.RunActionPrecommit(rs.Height, rs.Round) scheduleNextAction() continue ACTION_LOOP case RoundActionTryCommit: if rs.Step >= RoundStepCommit { continue ACTION_LOOP } if rs.Precommits.HasTwoThirdsMajority() { // Enter RoundStepCommit and commit. cs.RunActionCommit(rs.Height) continue ACTION_LOOP } else { // Could not commit, move onto next round. cs.SetupNewRound(rs.Height, rs.Round+1) // cs.Step is now at RoundStepNewRound scheduleNextAction() continue ACTION_LOOP } case RoundActionCommit: if rs.Step >= RoundStepCommit { continue ACTION_LOOP } // Enter RoundStepCommit and commit. cs.RunActionCommit(rs.Height) continue ACTION_LOOP case RoundActionTryFinalize: if cs.TryFinalizeCommit(rs.Height) { // Now at new height // cs.Step is at RoundStepNewHeight or RoundStepNewRound. // fire some events! go func() { newBlock := cs.blockStore.LoadBlock(cs.state.LastBlockHeight) cs.evsw.FireEvent(types.EventStringNewBlock(), newBlock) cs.evc.Flush() }() scheduleNextAction() continue ACTION_LOOP } else { // do not schedule next action. continue ACTION_LOOP } default: panic("Unknown action") } // For clarity, ensure that all switch cases call "continue" panic("Should not happen.") } } // Updates ConsensusState and increments height to match that of state. // If calculated round is greater than 0 (based on BlockTime or calculated StartTime) // then also sets up the appropriate round, and cs.Step becomes RoundStepNewRound. // Otherwise the round is 0 and cs.Step becomes RoundStepNewHeight. func (cs *ConsensusState) updateToState(state *sm.State, contiguous bool) { // Sanity check state. if contiguous && cs.Height > 0 && cs.Height != state.LastBlockHeight { panic(Fmt("updateToState() expected state height of %v but found %v", cs.Height, state.LastBlockHeight)) } // Reset fields based on state. validators := state.BondedValidators height := state.LastBlockHeight + 1 // next desired block height // RoundState fields cs.Height = height cs.Round = 0 cs.Step = RoundStepNewHeight if cs.CommitTime.IsZero() { // "Now" makes it easier to sync up dev nodes. // We add newHeightDelta to allow transactions // to be gathered for the first block. // And alternative solution that relies on clocks: // cs.StartTime = state.LastBlockTime.Add(newHeightDelta) cs.StartTime = time.Now().Add(newHeightDelta) } else { cs.StartTime = cs.CommitTime.Add(newHeightDelta) } cs.CommitTime = time.Time{} cs.Validators = validators cs.Proposal = nil cs.ProposalBlock = nil cs.ProposalBlockParts = nil cs.ProposalPOL = nil cs.ProposalPOLParts = nil cs.LockedBlock = nil cs.LockedBlockParts = nil cs.LockedPOL = nil cs.Prevotes = NewVoteSet(height, 0, types.VoteTypePrevote, validators) cs.Precommits = NewVoteSet(height, 0, types.VoteTypePrecommit, validators) cs.LastCommits = cs.Commits cs.Commits = NewVoteSet(height, 0, types.VoteTypeCommit, validators) cs.state = state cs.stagedBlock = nil cs.stagedState = nil // Update the round if we need to. round := calcRound(cs.StartTime) if round > 0 { cs.setupNewRound(round) } // If we've timed out, then send rebond tx. if cs.privValidator != nil && cs.state.UnbondingValidators.HasAddress(cs.privValidator.Address) { rebondTx := &types.RebondTx{ Address: cs.privValidator.Address, Height: cs.Height + 1, } err := cs.privValidator.SignRebondTx(rebondTx) if err == nil { log.Info("Signed and broadcast RebondTx", "height", cs.Height, "round", cs.Round, "tx", rebondTx) cs.mempoolReactor.BroadcastTx(rebondTx) } else { log.Warn("Error signing RebondTx", "height", cs.Height, "round", cs.Round, "tx", rebondTx, "error", err) } } } // After the call cs.Step becomes RoundStepNewRound. func (cs *ConsensusState) setupNewRound(round uint) { // XXX Looks like this is just not called. // Sanity check if round == 0 { panic("setupNewRound() should never be called for round 0") } // Increment all the way to round. validators := cs.Validators.Copy() validators.IncrementAccum(round - cs.Round) cs.Round = round cs.Step = RoundStepNewRound cs.Validators = validators cs.Proposal = nil cs.ProposalBlock = nil cs.ProposalBlockParts = nil cs.ProposalPOL = nil cs.ProposalPOLParts = nil cs.Prevotes = NewVoteSet(cs.Height, round, types.VoteTypePrevote, validators) cs.Prevotes.AddFromCommits(cs.Commits) cs.Precommits = NewVoteSet(cs.Height, round, types.VoteTypePrecommit, validators) cs.Precommits.AddFromCommits(cs.Commits) } func (cs *ConsensusState) SetPrivValidator(priv *sm.PrivValidator) { cs.mtx.Lock() defer cs.mtx.Unlock() cs.privValidator = priv } //----------------------------------------------------------------------------- // Set up the round to desired round and set step to RoundStepNewRound func (cs *ConsensusState) SetupNewRound(height uint, desiredRound uint) bool { cs.mtx.Lock() defer cs.mtx.Unlock() if cs.Height != height { return false } if desiredRound <= cs.Round { return false } cs.setupNewRound(desiredRound) // c.Step is now RoundStepNewRound cs.newStepCh <- cs.getRoundState() return true } func (cs *ConsensusState) RunActionPropose(height uint, round uint) { cs.mtx.Lock() defer cs.mtx.Unlock() if cs.Height != height || cs.Round != round { return } defer func() { cs.Step = RoundStepPropose cs.newStepCh <- cs.getRoundState() }() // Nothing to do if it's not our turn. if cs.privValidator == nil { return } if !bytes.Equal(cs.Validators.Proposer().Address, cs.privValidator.Address) { log.Debug("Not our turn to propose", "proposer", cs.Validators.Proposer().Address, "privValidator", cs.privValidator) return } else { log.Debug("Our turn to propose", "proposer", cs.Validators.Proposer().Address, "privValidator", cs.privValidator) } var block *types.Block var blockParts *types.PartSet var pol *POL var polParts *types.PartSet // Decide on block and POL if cs.LockedBlock != nil { // If we're locked onto a block, just choose that. block = cs.LockedBlock blockParts = cs.LockedBlockParts pol = cs.LockedPOL } else { // Otherwise we should create a new proposal. var validation *types.Validation if cs.Height == 1 { // We're creating a proposal for the first block. // The validation is empty. validation = &types.Validation{} } else if cs.LastCommits.HasTwoThirdsMajority() { // Make the validation from LastCommits validation = cs.LastCommits.MakeValidation() } else { // Upon reboot, we may have to use SeenValidation validation = cs.blockStore.LoadSeenValidation(height - 1) if validation == nil { // We just don't have any validation for the previous block log.Debug("Cannot propose anything: No validation for the previous block.") return } } txs := cs.mempoolReactor.Mempool.GetProposalTxs() block = &types.Block{ Header: &types.Header{ Network: config.App().GetString("Network"), Height: cs.Height, Time: time.Now(), Fees: 0, // TODO fees NumTxs: uint(len(txs)), LastBlockHash: cs.state.LastBlockHash, LastBlockParts: cs.state.LastBlockParts, StateHash: nil, // Will set afterwards. }, Validation: validation, Data: &types.Data{ Txs: txs, }, } // Set the types.Header.StateHash. err := cs.state.SetBlockStateHash(block) if err != nil { log.Error("Error setting state hash", "error", err) return } blockParts = block.MakePartSet() pol = cs.LockedPOL // If exists, is a PoUnlock. } if pol != nil { polParts = pol.MakePartSet() } // Make proposal proposal := NewProposal(cs.Height, cs.Round, blockParts.Header(), polParts.Header()) err := cs.privValidator.SignProposal(proposal) if err == nil { log.Info("Signed and set proposal", "height", cs.Height, "round", cs.Round, "proposal", proposal) log.Debug(Fmt("Signed and set proposal block: %v", block)) // Set fields cs.Proposal = proposal cs.ProposalBlock = block cs.ProposalBlockParts = blockParts cs.ProposalPOL = pol cs.ProposalPOLParts = polParts } else { log.Warn("Error signing proposal", "height", cs.Height, "round", cs.Round, "error", err) } } // Prevote for LockedBlock if we're locked, or ProposealBlock if valid. // Otherwise vote nil. func (cs *ConsensusState) RunActionPrevote(height uint, round uint) { cs.mtx.Lock() defer cs.mtx.Unlock() if cs.Height != height || cs.Round != round { panic(Fmt("RunActionPrevote(%v/%v), expected %v/%v", height, round, cs.Height, cs.Round)) } defer func() { cs.Step = RoundStepPrevote cs.newStepCh <- cs.getRoundState() }() // If a block is locked, prevote that. if cs.LockedBlock != nil { cs.signAddVote(types.VoteTypePrevote, cs.LockedBlock.Hash(), cs.LockedBlockParts.Header()) return } // If ProposalBlock is nil, prevote nil. if cs.ProposalBlock == nil { log.Warn("ProposalBlock is nil") cs.signAddVote(types.VoteTypePrevote, nil, types.PartSetHeader{}) return } // Try staging cs.ProposalBlock err := cs.stageBlock(cs.ProposalBlock, cs.ProposalBlockParts) if err != nil { // ProposalBlock is invalid, prevote nil. log.Warn("ProposalBlock is invalid", "error", err) cs.signAddVote(types.VoteTypePrevote, nil, types.PartSetHeader{}) return } // Prevote cs.ProposalBlock cs.signAddVote(types.VoteTypePrevote, cs.ProposalBlock.Hash(), cs.ProposalBlockParts.Header()) return } // Lock & Precommit the ProposalBlock if we have enough prevotes for it, // or unlock an existing lock if +2/3 of prevotes were nil. func (cs *ConsensusState) RunActionPrecommit(height uint, round uint) { cs.mtx.Lock() defer cs.mtx.Unlock() if cs.Height != height || cs.Round != round { panic(Fmt("RunActionPrecommit(%v/%v), expected %v/%v", height, round, cs.Height, cs.Round)) } defer func() { cs.Step = RoundStepPrecommit cs.newStepCh <- cs.getRoundState() }() hash, partsHeader, ok := cs.Prevotes.TwoThirdsMajority() if !ok { // If we don't have two thirds of prevotes, // don't do anything at all. return } // Remember this POL. (hash may be nil) cs.LockedPOL = cs.Prevotes.MakePOL() // If +2/3 prevoted nil. Just unlock. if len(hash) == 0 { cs.LockedBlock = nil cs.LockedBlockParts = nil return } // If +2/3 prevoted for already locked block, precommit it. if cs.LockedBlock.HashesTo(hash) { cs.signAddVote(types.VoteTypePrecommit, hash, partsHeader) return } // If +2/3 prevoted for cs.ProposalBlock, lock it and precommit it. if cs.ProposalBlock.HashesTo(hash) { // Validate the block. if err := cs.stageBlock(cs.ProposalBlock, cs.ProposalBlockParts); err != nil { // Prevent zombies. log.Warn("+2/3 prevoted for an invalid block", "error", err) return } cs.LockedBlock = cs.ProposalBlock cs.LockedBlockParts = cs.ProposalBlockParts cs.signAddVote(types.VoteTypePrecommit, hash, partsHeader) return } // We don't have the block that validators prevoted. // Unlock if we're locked. cs.LockedBlock = nil cs.LockedBlockParts = nil return } // Enter commit step. See the diagram for details. // There are two ways to enter this step: // * After the Precommit step with +2/3 precommits, or, // * Upon +2/3 commits regardless of current step // Either way this action is run at most once per round. func (cs *ConsensusState) RunActionCommit(height uint) { cs.mtx.Lock() defer cs.mtx.Unlock() if cs.Height != height { panic(Fmt("RunActionCommit(%v), expected %v", height, cs.Height)) } defer func() { cs.Step = RoundStepCommit cs.newStepCh <- cs.getRoundState() }() // Sanity check. // There are two ways to enter: // 1. +2/3 precommits at the end of RoundStepPrecommit // 2. +2/3 commits at any time hash, partsHeader, ok := cs.Precommits.TwoThirdsMajority() if !ok { hash, partsHeader, ok = cs.Commits.TwoThirdsMajority() if !ok { panic("RunActionCommit() expects +2/3 precommits or commits") } } // Clear the Locked* fields and use cs.Proposed* if cs.LockedBlock.HashesTo(hash) { cs.ProposalBlock = cs.LockedBlock cs.ProposalBlockParts = cs.LockedBlockParts cs.LockedBlock = nil cs.LockedBlockParts = nil cs.LockedPOL = nil } // If we don't have the block being committed, set up to get it. if !cs.ProposalBlock.HashesTo(hash) { if !cs.ProposalBlockParts.HasHeader(partsHeader) { // We're getting the wrong block. // Set up ProposalBlockParts and keep waiting. cs.ProposalBlock = nil cs.ProposalBlockParts = types.NewPartSetFromHeader(partsHeader) } else { // We just need to keep waiting. } } else { // We have the block, so sign a Commit-vote. cs.commitVoteBlock(cs.ProposalBlock, cs.ProposalBlockParts) } // If we have the block AND +2/3 commits, queue RoundActionTryFinalize. // Round will immediately become finalized. if cs.ProposalBlock.HashesTo(hash) && cs.Commits.HasTwoThirdsMajority() { cs.queueAction(RoundAction{cs.Height, cs.Round, RoundActionTryFinalize}) } } // Returns true if Finalize happened, which increments height && sets // the step to RoundStepNewHeight (or RoundStepNewRound, but probably not). func (cs *ConsensusState) TryFinalizeCommit(height uint) bool { cs.mtx.Lock() defer cs.mtx.Unlock() if cs.Height != height { panic(Fmt("TryFinalizeCommit(%v), expected %v", height, cs.Height)) } if cs.Step == RoundStepCommit && cs.Commits.HasTwoThirdsMajority() && cs.ProposalBlockParts.IsComplete() { // Sanity check if cs.ProposalBlock == nil { panic(Fmt("Expected ProposalBlock to exist")) } hash, header, _ := cs.Commits.TwoThirdsMajority() if !cs.ProposalBlock.HashesTo(hash) { // XXX See: https://github.com/tendermint/tendermint/issues/44 panic(Fmt("Expected ProposalBlock to hash to commit hash. Expected %X, got %X", hash, cs.ProposalBlock.Hash())) } if !cs.ProposalBlockParts.HasHeader(header) { panic(Fmt("Expected ProposalBlockParts header to be commit header")) } err := cs.stageBlock(cs.ProposalBlock, cs.ProposalBlockParts) if err == nil { log.Debug(Fmt("Finalizing commit of block: %v", cs.ProposalBlock)) // We have the block, so save/stage/sign-commit-vote. cs.saveCommitVoteBlock(cs.ProposalBlock, cs.ProposalBlockParts, cs.Commits) // Increment height. cs.updateToState(cs.stagedState, true) // cs.Step is now RoundStepNewHeight or RoundStepNewRound cs.newStepCh <- cs.getRoundState() return true } else { // Prevent zombies. // TODO: Does this ever happen? panic(Fmt("+2/3 committed an invalid block: %v", err)) } } return false } //----------------------------------------------------------------------------- func (cs *ConsensusState) SetProposal(proposal *Proposal) error { cs.mtx.Lock() defer cs.mtx.Unlock() // Already have one 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 cs.Step == RoundStepCommit { return nil } // Verify signature if !cs.Validators.Proposer().PubKey.VerifyBytes(account.SignBytes(proposal), proposal.Signature) { return ErrInvalidProposalSignature } cs.Proposal = proposal cs.ProposalBlockParts = types.NewPartSetFromHeader(proposal.BlockParts) cs.ProposalPOLParts = types.NewPartSetFromHeader(proposal.POLParts) return nil } // NOTE: block is not necessarily valid. // NOTE: This function may increment the height. func (cs *ConsensusState) AddProposalBlockPart(height uint, round uint, part *types.Part) (added bool, err error) { cs.mtx.Lock() defer cs.mtx.Unlock() // 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) if err != nil { return added, err } if added && cs.ProposalBlockParts.IsComplete() { var n int64 var err error cs.ProposalBlock = binary.ReadBinary(&types.Block{}, cs.ProposalBlockParts.GetReader(), &n, &err).(*types.Block) // If we're already in the commit step, try to finalize round. if cs.Step == RoundStepCommit { cs.queueAction(RoundAction{cs.Height, cs.Round, RoundActionTryFinalize}) } // XXX If POL is valid, consider unlocking. return true, err } return true, nil } // NOTE: POL is not necessarily valid. func (cs *ConsensusState) AddProposalPOLPart(height uint, round uint, part *types.Part) (added bool, err error) { cs.mtx.Lock() defer cs.mtx.Unlock() if cs.Height != height || cs.Round != round { return false, nil } // We're not expecting a POL part. if cs.ProposalPOLParts == nil { return false, nil // TODO: bad peer? Return error? } added, err = cs.ProposalPOLParts.AddPart(part) if err != nil { return added, err } if added && cs.ProposalPOLParts.IsComplete() { var n int64 var err error cs.ProposalPOL = binary.ReadBinary(&POL{}, cs.ProposalPOLParts.GetReader(), &n, &err).(*POL) return true, err } return true, nil } func (cs *ConsensusState) AddVote(address []byte, vote *types.Vote) (added bool, index uint, err error) { cs.mtx.Lock() defer cs.mtx.Unlock() return cs.addVote(address, vote) } //----------------------------------------------------------------------------- func (cs *ConsensusState) addVote(address []byte, vote *types.Vote) (added bool, index uint, err error) { switch vote.Type { case types.VoteTypePrevote: // Prevotes checks for height+round match. return cs.Prevotes.Add(address, vote) case types.VoteTypePrecommit: // Precommits checks for height+round match. return cs.Precommits.Add(address, vote) case types.VoteTypeCommit: if vote.Height == cs.Height { // No need to check if vote.Round < cs.Round ... // Prevotes && Precommits already checks that. cs.Prevotes.Add(address, vote) cs.Precommits.Add(address, vote) added, index, err = cs.Commits.Add(address, vote) if added && cs.Commits.HasTwoThirdsMajority() && cs.CommitTime.IsZero() { cs.CommitTime = time.Now() log.Debug(Fmt("Set CommitTime to %v", cs.CommitTime)) if cs.Step < RoundStepCommit { cs.queueAction(RoundAction{cs.Height, cs.Round, RoundActionCommit}) } else { cs.queueAction(RoundAction{cs.Height, cs.Round, RoundActionTryFinalize}) } } return added, index, err } if vote.Height+1 == cs.Height { return cs.LastCommits.Add(address, vote) } return false, 0, nil default: panic("Unknown vote type") } } func (cs *ConsensusState) stageBlock(block *types.Block, blockParts *types.PartSet) error { if block == nil { panic("Cannot stage nil block") } // Already staged? blockHash := block.Hash() if cs.stagedBlock != nil && len(blockHash) != 0 && bytes.Equal(cs.stagedBlock.Hash(), blockHash) { return nil } // Create a copy of the state for staging stateCopy := cs.state.Copy() // reset the event cache and pass it into the state cs.evc = events.NewEventCache(cs.evsw) stateCopy.SetFireable(cs.evc) // Commit block onto the copied state. // NOTE: Basic validation is done in state.AppendBlock(). err := sm.ExecBlock(stateCopy, block, blockParts.Header()) if err != nil { return err } else { cs.stagedBlock = block cs.stagedState = stateCopy return nil } } func (cs *ConsensusState) signAddVote(type_ byte, hash []byte, header types.PartSetHeader) *types.Vote { if cs.privValidator == nil || !cs.Validators.HasAddress(cs.privValidator.Address) { return nil } vote := &types.Vote{ Height: cs.Height, Round: cs.Round, Type: type_, BlockHash: hash, BlockParts: header, } err := cs.privValidator.SignVote(vote) if err == nil { log.Info("Signed and added vote", "height", cs.Height, "round", cs.Round, "vote", vote) cs.addVote(cs.privValidator.Address, vote) return vote } else { log.Warn("Error signing vote", "height", cs.Height, "round", cs.Round, "vote", vote, "error", err) return nil } } // sign a Commit-Vote func (cs *ConsensusState) commitVoteBlock(block *types.Block, blockParts *types.PartSet) { // The proposal must be valid. if err := cs.stageBlock(block, blockParts); err != nil { // Prevent zombies. log.Warn("commitVoteBlock() an invalid block", "error", err) return } // Commit-vote. if cs.lastCommitVoteHeight < block.Height { cs.signAddVote(types.VoteTypeCommit, block.Hash(), blockParts.Header()) cs.lastCommitVoteHeight = block.Height } else { log.Error("Duplicate commitVoteBlock() attempt", "lastCommitVoteHeight", cs.lastCommitVoteHeight, "types.Height", block.Height) } } // Save Block, save the +2/3 Commits we've seen, // and sign a Commit-Vote if we haven't already func (cs *ConsensusState) saveCommitVoteBlock(block *types.Block, blockParts *types.PartSet, commits *VoteSet) { // The proposal must be valid. if err := cs.stageBlock(block, blockParts); err != nil { // Prevent zombies. log.Warn("saveCommitVoteBlock() an invalid block", "error", err) return } // Save to blockStore. if cs.blockStore.Height() < block.Height { seenValidation := commits.MakeValidation() cs.blockStore.SaveBlock(block, blockParts, seenValidation) } // Save the state. cs.stagedState.Save() // Update mempool. cs.mempoolReactor.Mempool.ResetForBlockAndState(block, cs.stagedState) // Commit-vote if we haven't already. if cs.lastCommitVoteHeight < block.Height { cs.signAddVote(types.VoteTypeCommit, block.Hash(), blockParts.Header()) cs.lastCommitVoteHeight = block.Height } } // implements events.Eventable func (cs *ConsensusState) SetFireable(evsw events.Fireable) { cs.evsw = evsw } //----------------------------------------------------------------------------- // total duration of given round func calcRoundDuration(round uint) time.Duration { return RoundDuration0 + RoundDurationDelta*time.Duration(round) } // startTime is when round zero started. func calcRoundStartTime(round uint, startTime time.Time) time.Time { return startTime.Add(RoundDuration0*time.Duration(round) + RoundDurationDelta*(time.Duration((int64(round)*int64(round)-int64(round))/2))) } // calculates the current round given startTime of round zero. // NOTE: round is zero if startTime is in the future. func calcRound(startTime time.Time) uint { now := time.Now() if now.Before(startTime) { return 0 } // Start + D_0 * R + D_delta * (R^2 - R)/2 <= Now; find largest integer R. // D_delta * R^2 + (2D_0 - D_delta) * R + 2(Start - Now) <= 0. // AR^2 + BR + C <= 0; A = D_delta, B = (2_D0 - D_delta), C = 2(Start - Now). // R = Floor((-B + Sqrt(B^2 - 4AC))/2A) A := float64(RoundDurationDelta) B := 2.0*float64(RoundDuration0) - float64(RoundDurationDelta) C := 2.0 * float64(startTime.Sub(now)) R := math.Floor((-B + math.Sqrt(B*B-4.0*A*C)) / (2 * A)) if math.IsNaN(R) { panic("Could not calc round, should not happen") } if R > math.MaxInt32 { panic(Fmt("Could not calc round, round overflow: %v", R)) } if R < 0 { return 0 } return uint(R) } // convenience // NOTE: elapsedRatio can be negative if startTime is in the future. func calcRoundInfo(startTime time.Time) (round uint, roundStartTime time.Time, RoundDuration time.Duration, roundElapsed time.Duration, elapsedRatio float64) { round = calcRound(startTime) roundStartTime = calcRoundStartTime(round, startTime) RoundDuration = calcRoundDuration(round) roundElapsed = time.Now().Sub(roundStartTime) elapsedRatio = float64(roundElapsed) / float64(RoundDuration) return }