You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

1516 lines
51 KiB

10 years ago
8 years ago
10 years ago
10 years ago
10 years ago
10 years ago
9 years ago
10 years ago
10 years ago
10 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
10 years ago
8 years ago
8 years ago
10 years ago
10 years ago
8 years ago
10 years ago
9 years ago
10 years ago
8 years ago
10 years ago
10 years ago
10 years ago
8 years ago
8 years ago
10 years ago
8 years ago
8 years ago
9 years ago
8 years ago
9 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
10 years ago
8 years ago
10 years ago
8 years ago
10 years ago
8 years ago
10 years ago
10 years ago
9 years ago
10 years ago
8 years ago
8 years ago
10 years ago
8 years ago
8 years ago
10 years ago
10 years ago
10 years ago
8 years ago
8 years ago
10 years ago
10 years ago
10 years ago
8 years ago
10 years ago
8 years ago
8 years ago
8 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
8 years ago
10 years ago
10 years ago
10 years ago
8 years ago
8 years ago
8 years ago
10 years ago
8 years ago
10 years ago
10 years ago
8 years ago
8 years ago
10 years ago
10 years ago
8 years ago
8 years ago
10 years ago
10 years ago
8 years ago
10 years ago
10 years ago
8 years ago
10 years ago
10 years ago
10 years ago
8 years ago
8 years ago
9 years ago
8 years ago
10 years ago
8 years ago
9 years ago
9 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
9 years ago
10 years ago
10 years ago
8 years ago
10 years ago
8 years ago
10 years ago
8 years ago
10 years ago
10 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
9 years ago
9 years ago
8 years ago
9 years ago
8 years ago
8 years ago
8 years ago
8 years ago
10 years ago
8 years ago
10 years ago
10 years ago
8 years ago
8 years ago
10 years ago
8 years ago
8 years ago
8 years ago
10 years ago
  1. package consensus
  2. import (
  3. "bytes"
  4. "errors"
  5. "fmt"
  6. "path"
  7. "reflect"
  8. "sync"
  9. "time"
  10. fail "github.com/ebuchman/fail-test"
  11. wire "github.com/tendermint/go-wire"
  12. cfg "github.com/tendermint/tendermint/config"
  13. "github.com/tendermint/tendermint/proxy"
  14. sm "github.com/tendermint/tendermint/state"
  15. "github.com/tendermint/tendermint/types"
  16. cmn "github.com/tendermint/tmlibs/common"
  17. "github.com/tendermint/tmlibs/log"
  18. )
  19. //-----------------------------------------------------------------------------
  20. // Config
  21. //-----------------------------------------------------------------------------
  22. // Errors
  23. var (
  24. ErrInvalidProposalSignature = errors.New("Error invalid proposal signature")
  25. ErrInvalidProposalPOLRound = errors.New("Error invalid proposal POL round")
  26. ErrAddingVote = errors.New("Error adding vote")
  27. ErrVoteHeightMismatch = errors.New("Error vote height mismatch")
  28. )
  29. //-----------------------------------------------------------------------------
  30. // RoundStepType enum type
  31. type RoundStepType uint8 // These must be numeric, ordered.
  32. const (
  33. RoundStepNewHeight = RoundStepType(0x01) // Wait til CommitTime + timeoutCommit
  34. RoundStepNewRound = RoundStepType(0x02) // Setup new round and go to RoundStepPropose
  35. RoundStepPropose = RoundStepType(0x03) // Did propose, gossip proposal
  36. RoundStepPrevote = RoundStepType(0x04) // Did prevote, gossip prevotes
  37. RoundStepPrevoteWait = RoundStepType(0x05) // Did receive any +2/3 prevotes, start timeout
  38. RoundStepPrecommit = RoundStepType(0x06) // Did precommit, gossip precommits
  39. RoundStepPrecommitWait = RoundStepType(0x07) // Did receive any +2/3 precommits, start timeout
  40. RoundStepCommit = RoundStepType(0x08) // Entered commit state machine
  41. // NOTE: RoundStepNewHeight acts as RoundStepCommitWait.
  42. )
  43. func (rs RoundStepType) String() string {
  44. switch rs {
  45. case RoundStepNewHeight:
  46. return "RoundStepNewHeight"
  47. case RoundStepNewRound:
  48. return "RoundStepNewRound"
  49. case RoundStepPropose:
  50. return "RoundStepPropose"
  51. case RoundStepPrevote:
  52. return "RoundStepPrevote"
  53. case RoundStepPrevoteWait:
  54. return "RoundStepPrevoteWait"
  55. case RoundStepPrecommit:
  56. return "RoundStepPrecommit"
  57. case RoundStepPrecommitWait:
  58. return "RoundStepPrecommitWait"
  59. case RoundStepCommit:
  60. return "RoundStepCommit"
  61. default:
  62. return "RoundStepUnknown" // Cannot panic.
  63. }
  64. }
  65. //-----------------------------------------------------------------------------
  66. // Immutable when returned from ConsensusState.GetRoundState()
  67. // TODO: Actually, only the top pointer is copied,
  68. // so access to field pointers is still racey
  69. type RoundState struct {
  70. Height int // Height we are working on
  71. Round int
  72. Step RoundStepType
  73. StartTime time.Time
  74. CommitTime time.Time // Subjective time when +2/3 precommits for Block at Round were found
  75. Validators *types.ValidatorSet
  76. Proposal *types.Proposal
  77. ProposalBlock *types.Block
  78. ProposalBlockParts *types.PartSet
  79. LockedRound int
  80. LockedBlock *types.Block
  81. LockedBlockParts *types.PartSet
  82. Votes *HeightVoteSet
  83. CommitRound int //
  84. LastCommit *types.VoteSet // Last precommits at Height-1
  85. LastValidators *types.ValidatorSet
  86. }
  87. func (rs *RoundState) RoundStateEvent() types.EventDataRoundState {
  88. edrs := types.EventDataRoundState{
  89. Height: rs.Height,
  90. Round: rs.Round,
  91. Step: rs.Step.String(),
  92. RoundState: rs,
  93. }
  94. return edrs
  95. }
  96. func (rs *RoundState) String() string {
  97. return rs.StringIndented("")
  98. }
  99. func (rs *RoundState) StringIndented(indent string) string {
  100. return fmt.Sprintf(`RoundState{
  101. %s H:%v R:%v S:%v
  102. %s StartTime: %v
  103. %s CommitTime: %v
  104. %s Validators: %v
  105. %s Proposal: %v
  106. %s ProposalBlock: %v %v
  107. %s LockedRound: %v
  108. %s LockedBlock: %v %v
  109. %s Votes: %v
  110. %s LastCommit: %v
  111. %s LastValidators: %v
  112. %s}`,
  113. indent, rs.Height, rs.Round, rs.Step,
  114. indent, rs.StartTime,
  115. indent, rs.CommitTime,
  116. indent, rs.Validators.StringIndented(indent+" "),
  117. indent, rs.Proposal,
  118. indent, rs.ProposalBlockParts.StringShort(), rs.ProposalBlock.StringShort(),
  119. indent, rs.LockedRound,
  120. indent, rs.LockedBlockParts.StringShort(), rs.LockedBlock.StringShort(),
  121. indent, rs.Votes.StringIndented(indent+" "),
  122. indent, rs.LastCommit.StringShort(),
  123. indent, rs.LastValidators.StringIndented(indent+" "),
  124. indent)
  125. }
  126. func (rs *RoundState) StringShort() string {
  127. return fmt.Sprintf(`RoundState{H:%v R:%v S:%v ST:%v}`,
  128. rs.Height, rs.Round, rs.Step, rs.StartTime)
  129. }
  130. //-----------------------------------------------------------------------------
  131. var (
  132. msgQueueSize = 1000
  133. )
  134. // msgs from the reactor which may update the state
  135. type msgInfo struct {
  136. Msg ConsensusMessage `json:"msg"`
  137. PeerKey string `json:"peer_key"`
  138. }
  139. // internally generated messages which may update the state
  140. type timeoutInfo struct {
  141. Duration time.Duration `json:"duration"`
  142. Height int `json:"height"`
  143. Round int `json:"round"`
  144. Step RoundStepType `json:"step"`
  145. }
  146. func (ti *timeoutInfo) String() string {
  147. return fmt.Sprintf("%v ; %d/%d %v", ti.Duration, ti.Height, ti.Round, ti.Step)
  148. }
  149. type PrivValidator interface {
  150. GetAddress() []byte
  151. SignVote(chainID string, vote *types.Vote) error
  152. SignProposal(chainID string, proposal *types.Proposal) error
  153. }
  154. // Tracks consensus state across block heights and rounds.
  155. type ConsensusState struct {
  156. cmn.BaseService
  157. // config details
  158. config *cfg.ConsensusConfig
  159. privValidator PrivValidator // for signing votes
  160. // services for creating and executing blocks
  161. proxyAppConn proxy.AppConnConsensus
  162. blockStore types.BlockStore
  163. mempool types.Mempool
  164. // internal state
  165. mtx sync.Mutex
  166. RoundState
  167. state *sm.State // State until height-1.
  168. // state changes may be triggered by msgs from peers,
  169. // msgs from ourself, or by timeouts
  170. peerMsgQueue chan msgInfo
  171. internalMsgQueue chan msgInfo
  172. timeoutTicker TimeoutTicker
  173. // we use PubSub to trigger msg broadcasts in the reactor,
  174. // and to notify external subscribers, eg. through a websocket
  175. evsw types.EventSwitch
  176. // a Write-Ahead Log ensures we can recover from any kind of crash
  177. // and helps us avoid signing conflicting votes
  178. wal *WAL
  179. replayMode bool // so we don't log signing errors during replay
  180. // for tests where we want to limit the number of transitions the state makes
  181. nSteps int
  182. // some functions can be overwritten for testing
  183. decideProposal func(height, round int)
  184. doPrevote func(height, round int)
  185. setProposal func(proposal *types.Proposal) error
  186. // closed when we finish shutting down
  187. done chan struct{}
  188. }
  189. func NewConsensusState(config *cfg.ConsensusConfig, state *sm.State, proxyAppConn proxy.AppConnConsensus, blockStore types.BlockStore, mempool types.Mempool) *ConsensusState {
  190. cs := &ConsensusState{
  191. config: config,
  192. proxyAppConn: proxyAppConn,
  193. blockStore: blockStore,
  194. mempool: mempool,
  195. peerMsgQueue: make(chan msgInfo, msgQueueSize),
  196. internalMsgQueue: make(chan msgInfo, msgQueueSize),
  197. timeoutTicker: NewTimeoutTicker(),
  198. done: make(chan struct{}),
  199. }
  200. // set function defaults (may be overwritten before calling Start)
  201. cs.decideProposal = cs.defaultDecideProposal
  202. cs.doPrevote = cs.defaultDoPrevote
  203. cs.setProposal = cs.defaultSetProposal
  204. cs.updateToState(state)
  205. // Don't call scheduleRound0 yet.
  206. // We do that upon Start().
  207. cs.reconstructLastCommit(state)
  208. cs.BaseService = *cmn.NewBaseService(nil, "ConsensusState", cs)
  209. return cs
  210. }
  211. //----------------------------------------
  212. // Public interface
  213. // SetLogger implements Service.
  214. func (cs *ConsensusState) SetLogger(l log.Logger) {
  215. cs.BaseService.Logger = l
  216. cs.timeoutTicker.SetLogger(l)
  217. }
  218. // SetEventSwitch implements events.Eventable
  219. func (cs *ConsensusState) SetEventSwitch(evsw types.EventSwitch) {
  220. cs.evsw = evsw
  221. }
  222. func (cs *ConsensusState) String() string {
  223. // better not to access shared variables
  224. return cmn.Fmt("ConsensusState") //(H:%v R:%v S:%v", cs.Height, cs.Round, cs.Step)
  225. }
  226. func (cs *ConsensusState) GetState() *sm.State {
  227. cs.mtx.Lock()
  228. defer cs.mtx.Unlock()
  229. return cs.state.Copy()
  230. }
  231. func (cs *ConsensusState) GetRoundState() *RoundState {
  232. cs.mtx.Lock()
  233. defer cs.mtx.Unlock()
  234. return cs.getRoundState()
  235. }
  236. func (cs *ConsensusState) getRoundState() *RoundState {
  237. rs := cs.RoundState // copy
  238. return &rs
  239. }
  240. func (cs *ConsensusState) GetValidators() (int, []*types.Validator) {
  241. cs.mtx.Lock()
  242. defer cs.mtx.Unlock()
  243. return cs.state.LastBlockHeight, cs.state.Validators.Copy().Validators
  244. }
  245. // Sets our private validator account for signing votes.
  246. func (cs *ConsensusState) SetPrivValidator(priv PrivValidator) {
  247. cs.mtx.Lock()
  248. defer cs.mtx.Unlock()
  249. cs.privValidator = priv
  250. }
  251. // Set the local timer
  252. func (cs *ConsensusState) SetTimeoutTicker(timeoutTicker TimeoutTicker) {
  253. cs.mtx.Lock()
  254. defer cs.mtx.Unlock()
  255. cs.timeoutTicker = timeoutTicker
  256. }
  257. func (cs *ConsensusState) LoadCommit(height int) *types.Commit {
  258. cs.mtx.Lock()
  259. defer cs.mtx.Unlock()
  260. if height == cs.blockStore.Height() {
  261. return cs.blockStore.LoadSeenCommit(height)
  262. }
  263. return cs.blockStore.LoadBlockCommit(height)
  264. }
  265. func (cs *ConsensusState) OnStart() error {
  266. walFile := cs.config.WalFile()
  267. if err := cs.OpenWAL(walFile); err != nil {
  268. cs.Logger.Error("Error loading ConsensusState wal", "error", err.Error())
  269. return err
  270. }
  271. // we need the timeoutRoutine for replay so
  272. // we don't block on the tick chan.
  273. // NOTE: we will get a build up of garbage go routines
  274. // firing on the tockChan until the receiveRoutine is started
  275. // to deal with them (by that point, at most one will be valid)
  276. cs.timeoutTicker.Start()
  277. // we may have lost some votes if the process crashed
  278. // reload from consensus log to catchup
  279. if err := cs.catchupReplay(cs.Height); err != nil {
  280. cs.Logger.Error("Error on catchup replay. Proceeding to start ConsensusState anyway", "error", err.Error())
  281. // NOTE: if we ever do return an error here,
  282. // make sure to stop the timeoutTicker
  283. }
  284. // now start the receiveRoutine
  285. go cs.receiveRoutine(0)
  286. // schedule the first round!
  287. // use GetRoundState so we don't race the receiveRoutine for access
  288. cs.scheduleRound0(cs.GetRoundState())
  289. return nil
  290. }
  291. // timeoutRoutine: receive requests for timeouts on tickChan and fire timeouts on tockChan
  292. // receiveRoutine: serializes processing of proposoals, block parts, votes; coordinates state transitions
  293. func (cs *ConsensusState) startRoutines(maxSteps int) {
  294. cs.timeoutTicker.Start()
  295. go cs.receiveRoutine(maxSteps)
  296. }
  297. func (cs *ConsensusState) OnStop() {
  298. cs.BaseService.OnStop()
  299. cs.timeoutTicker.Stop()
  300. // Make BaseService.Wait() wait until cs.wal.Wait()
  301. if cs.wal != nil && cs.IsRunning() {
  302. cs.wal.Wait()
  303. }
  304. }
  305. // NOTE: be sure to Stop() the event switch and drain
  306. // any event channels or this may deadlock
  307. func (cs *ConsensusState) Wait() {
  308. <-cs.done
  309. }
  310. // Open file to log all consensus messages and timeouts for deterministic accountability
  311. func (cs *ConsensusState) OpenWAL(walFile string) (err error) {
  312. err = cmn.EnsureDir(path.Dir(walFile), 0700)
  313. if err != nil {
  314. cs.Logger.Error("Error ensuring ConsensusState wal dir", "error", err.Error())
  315. return err
  316. }
  317. cs.mtx.Lock()
  318. defer cs.mtx.Unlock()
  319. wal, err := NewWAL(walFile, cs.config.WalLight)
  320. if err != nil {
  321. return err
  322. }
  323. wal.SetLogger(cs.Logger.With("wal", walFile))
  324. if _, err := wal.Start(); err != nil {
  325. return err
  326. }
  327. cs.wal = wal
  328. return nil
  329. }
  330. //------------------------------------------------------------
  331. // Public interface for passing messages into the consensus state,
  332. // possibly causing a state transition
  333. // TODO: should these return anything or let callers just use events?
  334. // May block on send if queue is full.
  335. func (cs *ConsensusState) AddVote(vote *types.Vote, peerKey string) (added bool, err error) {
  336. if peerKey == "" {
  337. cs.internalMsgQueue <- msgInfo{&VoteMessage{vote}, ""}
  338. } else {
  339. cs.peerMsgQueue <- msgInfo{&VoteMessage{vote}, peerKey}
  340. }
  341. // TODO: wait for event?!
  342. return false, nil
  343. }
  344. // May block on send if queue is full.
  345. func (cs *ConsensusState) SetProposal(proposal *types.Proposal, peerKey string) error {
  346. if peerKey == "" {
  347. cs.internalMsgQueue <- msgInfo{&ProposalMessage{proposal}, ""}
  348. } else {
  349. cs.peerMsgQueue <- msgInfo{&ProposalMessage{proposal}, peerKey}
  350. }
  351. // TODO: wait for event?!
  352. return nil
  353. }
  354. // May block on send if queue is full.
  355. func (cs *ConsensusState) AddProposalBlockPart(height, round int, part *types.Part, peerKey string) error {
  356. if peerKey == "" {
  357. cs.internalMsgQueue <- msgInfo{&BlockPartMessage{height, round, part}, ""}
  358. } else {
  359. cs.peerMsgQueue <- msgInfo{&BlockPartMessage{height, round, part}, peerKey}
  360. }
  361. // TODO: wait for event?!
  362. return nil
  363. }
  364. // May block on send if queue is full.
  365. func (cs *ConsensusState) SetProposalAndBlock(proposal *types.Proposal, block *types.Block, parts *types.PartSet, peerKey string) error {
  366. cs.SetProposal(proposal, peerKey)
  367. for i := 0; i < parts.Total(); i++ {
  368. part := parts.GetPart(i)
  369. cs.AddProposalBlockPart(proposal.Height, proposal.Round, part, peerKey)
  370. }
  371. return nil // TODO errors
  372. }
  373. //------------------------------------------------------------
  374. // internal functions for managing the state
  375. func (cs *ConsensusState) updateHeight(height int) {
  376. cs.Height = height
  377. }
  378. func (cs *ConsensusState) updateRoundStep(round int, step RoundStepType) {
  379. cs.Round = round
  380. cs.Step = step
  381. }
  382. // enterNewRound(height, 0) at cs.StartTime.
  383. func (cs *ConsensusState) scheduleRound0(rs *RoundState) {
  384. //cs.Logger.Info("scheduleRound0", "now", time.Now(), "startTime", cs.StartTime)
  385. sleepDuration := rs.StartTime.Sub(time.Now())
  386. cs.scheduleTimeout(sleepDuration, rs.Height, 0, RoundStepNewHeight)
  387. }
  388. // Attempt to schedule a timeout (by sending timeoutInfo on the tickChan)
  389. func (cs *ConsensusState) scheduleTimeout(duration time.Duration, height, round int, step RoundStepType) {
  390. cs.timeoutTicker.ScheduleTimeout(timeoutInfo{duration, height, round, step})
  391. }
  392. // send a msg into the receiveRoutine regarding our own proposal, block part, or vote
  393. func (cs *ConsensusState) sendInternalMessage(mi msgInfo) {
  394. select {
  395. case cs.internalMsgQueue <- mi:
  396. default:
  397. // NOTE: using the go-routine means our votes can
  398. // be processed out of order.
  399. // TODO: use CList here for strict determinism and
  400. // attempt push to internalMsgQueue in receiveRoutine
  401. cs.Logger.Info("Internal msg queue is full. Using a go-routine")
  402. go func() { cs.internalMsgQueue <- mi }()
  403. }
  404. }
  405. // Reconstruct LastCommit from SeenCommit, which we saved along with the block,
  406. // (which happens even before saving the state)
  407. func (cs *ConsensusState) reconstructLastCommit(state *sm.State) {
  408. if state.LastBlockHeight == 0 {
  409. return
  410. }
  411. seenCommit := cs.blockStore.LoadSeenCommit(state.LastBlockHeight)
  412. lastPrecommits := types.NewVoteSet(cs.state.ChainID, state.LastBlockHeight, seenCommit.Round(), types.VoteTypePrecommit, state.LastValidators)
  413. for _, precommit := range seenCommit.Precommits {
  414. if precommit == nil {
  415. continue
  416. }
  417. added, err := lastPrecommits.AddVote(precommit)
  418. if !added || err != nil {
  419. cmn.PanicCrisis(cmn.Fmt("Failed to reconstruct LastCommit: %v", err))
  420. }
  421. }
  422. if !lastPrecommits.HasTwoThirdsMajority() {
  423. cmn.PanicSanity("Failed to reconstruct LastCommit: Does not have +2/3 maj")
  424. }
  425. cs.LastCommit = lastPrecommits
  426. }
  427. // Updates ConsensusState and increments height to match that of state.
  428. // The round becomes 0 and cs.Step becomes RoundStepNewHeight.
  429. func (cs *ConsensusState) updateToState(state *sm.State) {
  430. if cs.CommitRound > -1 && 0 < cs.Height && cs.Height != state.LastBlockHeight {
  431. cmn.PanicSanity(cmn.Fmt("updateToState() expected state height of %v but found %v",
  432. cs.Height, state.LastBlockHeight))
  433. }
  434. if cs.state != nil && cs.state.LastBlockHeight+1 != cs.Height {
  435. // This might happen when someone else is mutating cs.state.
  436. // Someone forgot to pass in state.Copy() somewhere?!
  437. cmn.PanicSanity(cmn.Fmt("Inconsistent cs.state.LastBlockHeight+1 %v vs cs.Height %v",
  438. cs.state.LastBlockHeight+1, cs.Height))
  439. }
  440. // If state isn't further out than cs.state, just ignore.
  441. // This happens when SwitchToConsensus() is called in the reactor.
  442. // We don't want to reset e.g. the Votes.
  443. if cs.state != nil && (state.LastBlockHeight <= cs.state.LastBlockHeight) {
  444. cs.Logger.Info("Ignoring updateToState()", "newHeight", state.LastBlockHeight+1, "oldHeight", cs.state.LastBlockHeight+1)
  445. return
  446. }
  447. // Reset fields based on state.
  448. validators := state.Validators
  449. lastPrecommits := (*types.VoteSet)(nil)
  450. if cs.CommitRound > -1 && cs.Votes != nil {
  451. if !cs.Votes.Precommits(cs.CommitRound).HasTwoThirdsMajority() {
  452. cmn.PanicSanity("updateToState(state) called but last Precommit round didn't have +2/3")
  453. }
  454. lastPrecommits = cs.Votes.Precommits(cs.CommitRound)
  455. }
  456. // Next desired block height
  457. height := state.LastBlockHeight + 1
  458. // RoundState fields
  459. cs.updateHeight(height)
  460. cs.updateRoundStep(0, RoundStepNewHeight)
  461. if cs.CommitTime.IsZero() {
  462. // "Now" makes it easier to sync up dev nodes.
  463. // We add timeoutCommit to allow transactions
  464. // to be gathered for the first block.
  465. // And alternative solution that relies on clocks:
  466. // cs.StartTime = state.LastBlockTime.Add(timeoutCommit)
  467. cs.StartTime = cs.config.Commit(time.Now())
  468. } else {
  469. cs.StartTime = cs.config.Commit(cs.CommitTime)
  470. }
  471. cs.Validators = validators
  472. cs.Proposal = nil
  473. cs.ProposalBlock = nil
  474. cs.ProposalBlockParts = nil
  475. cs.LockedRound = 0
  476. cs.LockedBlock = nil
  477. cs.LockedBlockParts = nil
  478. cs.Votes = NewHeightVoteSet(state.ChainID, height, validators)
  479. cs.CommitRound = -1
  480. cs.LastCommit = lastPrecommits
  481. cs.LastValidators = state.LastValidators
  482. cs.state = state
  483. // Finally, broadcast RoundState
  484. cs.newStep()
  485. }
  486. func (cs *ConsensusState) newStep() {
  487. rs := cs.RoundStateEvent()
  488. cs.wal.Save(rs)
  489. cs.nSteps += 1
  490. // newStep is called by updateToStep in NewConsensusState before the evsw is set!
  491. if cs.evsw != nil {
  492. types.FireEventNewRoundStep(cs.evsw, rs)
  493. }
  494. }
  495. //-----------------------------------------
  496. // the main go routines
  497. // receiveRoutine handles messages which may cause state transitions.
  498. // it's argument (n) is the number of messages to process before exiting - use 0 to run forever
  499. // It keeps the RoundState and is the only thing that updates it.
  500. // Updates (state transitions) happen on timeouts, complete proposals, and 2/3 majorities
  501. func (cs *ConsensusState) receiveRoutine(maxSteps int) {
  502. for {
  503. if maxSteps > 0 {
  504. if cs.nSteps >= maxSteps {
  505. cs.Logger.Info("reached max steps. exiting receive routine")
  506. cs.nSteps = 0
  507. return
  508. }
  509. }
  510. rs := cs.RoundState
  511. var mi msgInfo
  512. select {
  513. case mi = <-cs.peerMsgQueue:
  514. cs.wal.Save(mi)
  515. // handles proposals, block parts, votes
  516. // may generate internal events (votes, complete proposals, 2/3 majorities)
  517. cs.handleMsg(mi, rs)
  518. case mi = <-cs.internalMsgQueue:
  519. cs.wal.Save(mi)
  520. // handles proposals, block parts, votes
  521. cs.handleMsg(mi, rs)
  522. case ti := <-cs.timeoutTicker.Chan(): // tockChan:
  523. cs.wal.Save(ti)
  524. // if the timeout is relevant to the rs
  525. // go to the next step
  526. cs.handleTimeout(ti, rs)
  527. case <-cs.Quit:
  528. // NOTE: the internalMsgQueue may have signed messages from our
  529. // priv_val that haven't hit the WAL, but its ok because
  530. // priv_val tracks LastSig
  531. // close wal now that we're done writing to it
  532. if cs.wal != nil {
  533. cs.wal.Stop()
  534. }
  535. close(cs.done)
  536. return
  537. }
  538. }
  539. }
  540. // state transitions on complete-proposal, 2/3-any, 2/3-one
  541. func (cs *ConsensusState) handleMsg(mi msgInfo, rs RoundState) {
  542. cs.mtx.Lock()
  543. defer cs.mtx.Unlock()
  544. var err error
  545. msg, peerKey := mi.Msg, mi.PeerKey
  546. switch msg := msg.(type) {
  547. case *ProposalMessage:
  548. // will not cause transition.
  549. // once proposal is set, we can receive block parts
  550. err = cs.setProposal(msg.Proposal)
  551. case *BlockPartMessage:
  552. // if the proposal is complete, we'll enterPrevote or tryFinalizeCommit
  553. _, err = cs.addProposalBlockPart(msg.Height, msg.Part, peerKey != "")
  554. if err != nil && msg.Round != cs.Round {
  555. err = nil
  556. }
  557. case *VoteMessage:
  558. // attempt to add the vote and dupeout the validator if its a duplicate signature
  559. // if the vote gives us a 2/3-any or 2/3-one, we transition
  560. err := cs.tryAddVote(msg.Vote, peerKey)
  561. if err == ErrAddingVote {
  562. // TODO: punish peer
  563. }
  564. // NOTE: the vote is broadcast to peers by the reactor listening
  565. // for vote events
  566. // TODO: If rs.Height == vote.Height && rs.Round < vote.Round,
  567. // the peer is sending us CatchupCommit precommits.
  568. // We could make note of this and help filter in broadcastHasVoteMessage().
  569. default:
  570. cs.Logger.Error("Unknown msg type", reflect.TypeOf(msg))
  571. }
  572. if err != nil {
  573. cs.Logger.Error("Error with msg", "type", reflect.TypeOf(msg), "peer", peerKey, "error", err, "msg", msg)
  574. }
  575. }
  576. func (cs *ConsensusState) handleTimeout(ti timeoutInfo, rs RoundState) {
  577. cs.Logger.Debug("Received tock", "timeout", ti.Duration, "height", ti.Height, "round", ti.Round, "step", ti.Step)
  578. // timeouts must be for current height, round, step
  579. if ti.Height != rs.Height || ti.Round < rs.Round || (ti.Round == rs.Round && ti.Step < rs.Step) {
  580. cs.Logger.Debug("Ignoring tock because we're ahead", "height", rs.Height, "round", rs.Round, "step", rs.Step)
  581. return
  582. }
  583. // the timeout will now cause a state transition
  584. cs.mtx.Lock()
  585. defer cs.mtx.Unlock()
  586. switch ti.Step {
  587. case RoundStepNewHeight:
  588. // NewRound event fired from enterNewRound.
  589. // XXX: should we fire timeout here (for timeout commit)?
  590. cs.enterNewRound(ti.Height, 0)
  591. case RoundStepPropose:
  592. types.FireEventTimeoutPropose(cs.evsw, cs.RoundStateEvent())
  593. cs.enterPrevote(ti.Height, ti.Round)
  594. case RoundStepPrevoteWait:
  595. types.FireEventTimeoutWait(cs.evsw, cs.RoundStateEvent())
  596. cs.enterPrecommit(ti.Height, ti.Round)
  597. case RoundStepPrecommitWait:
  598. types.FireEventTimeoutWait(cs.evsw, cs.RoundStateEvent())
  599. cs.enterNewRound(ti.Height, ti.Round+1)
  600. default:
  601. panic(cmn.Fmt("Invalid timeout step: %v", ti.Step))
  602. }
  603. }
  604. //-----------------------------------------------------------------------------
  605. // State functions
  606. // Used internally by handleTimeout and handleMsg to make state transitions
  607. // Enter: +2/3 precommits for nil at (height,round-1)
  608. // Enter: `timeoutPrecommits` after any +2/3 precommits from (height,round-1)
  609. // Enter: `startTime = commitTime+timeoutCommit` from NewHeight(height)
  610. // NOTE: cs.StartTime was already set for height.
  611. func (cs *ConsensusState) enterNewRound(height int, round int) {
  612. if cs.Height != height || round < cs.Round || (cs.Round == round && cs.Step != RoundStepNewHeight) {
  613. cs.Logger.Debug(cmn.Fmt("enterNewRound(%v/%v): Invalid args. Current step: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step))
  614. return
  615. }
  616. if now := time.Now(); cs.StartTime.After(now) {
  617. cs.Logger.Info("Need to set a buffer and log message here for sanity.", "startTime", cs.StartTime, "now", now)
  618. }
  619. cs.Logger.Info(cmn.Fmt("enterNewRound(%v/%v). Current: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step))
  620. // Increment validators if necessary
  621. validators := cs.Validators
  622. if cs.Round < round {
  623. validators = validators.Copy()
  624. validators.IncrementAccum(round - cs.Round)
  625. }
  626. // Setup new round
  627. // we don't fire newStep for this step,
  628. // but we fire an event, so update the round step first
  629. cs.updateRoundStep(round, RoundStepNewRound)
  630. cs.Validators = validators
  631. if round == 0 {
  632. // We've already reset these upon new height,
  633. // and meanwhile we might have received a proposal
  634. // for round 0.
  635. } else {
  636. cs.Proposal = nil
  637. cs.ProposalBlock = nil
  638. cs.ProposalBlockParts = nil
  639. }
  640. cs.Votes.SetRound(round + 1) // also track next round (round+1) to allow round-skipping
  641. types.FireEventNewRound(cs.evsw, cs.RoundStateEvent())
  642. // Immediately go to enterPropose.
  643. cs.enterPropose(height, round)
  644. }
  645. // Enter: from NewRound(height,round).
  646. func (cs *ConsensusState) enterPropose(height int, round int) {
  647. if cs.Height != height || round < cs.Round || (cs.Round == round && RoundStepPropose <= cs.Step) {
  648. cs.Logger.Debug(cmn.Fmt("enterPropose(%v/%v): Invalid args. Current step: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step))
  649. return
  650. }
  651. cs.Logger.Info(cmn.Fmt("enterPropose(%v/%v). Current: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step))
  652. defer func() {
  653. // Done enterPropose:
  654. cs.updateRoundStep(round, RoundStepPropose)
  655. cs.newStep()
  656. // If we have the whole proposal + POL, then goto Prevote now.
  657. // else, we'll enterPrevote when the rest of the proposal is received (in AddProposalBlockPart),
  658. // or else after timeoutPropose
  659. if cs.isProposalComplete() {
  660. cs.enterPrevote(height, cs.Round)
  661. }
  662. }()
  663. // If we don't get the proposal and all block parts quick enough, enterPrevote
  664. cs.scheduleTimeout(cs.config.Propose(round), height, round, RoundStepPropose)
  665. // Nothing more to do if we're not a validator
  666. if cs.privValidator == nil {
  667. return
  668. }
  669. if !bytes.Equal(cs.Validators.GetProposer().Address, cs.privValidator.GetAddress()) {
  670. cs.Logger.Info("enterPropose: Not our turn to propose", "proposer", cs.Validators.GetProposer().Address, "privValidator", cs.privValidator)
  671. } else {
  672. cs.Logger.Info("enterPropose: Our turn to propose", "proposer", cs.Validators.GetProposer().Address, "privValidator", cs.privValidator)
  673. cs.decideProposal(height, round)
  674. }
  675. }
  676. func (cs *ConsensusState) defaultDecideProposal(height, round int) {
  677. var block *types.Block
  678. var blockParts *types.PartSet
  679. // Decide on block
  680. if cs.LockedBlock != nil {
  681. // If we're locked onto a block, just choose that.
  682. block, blockParts = cs.LockedBlock, cs.LockedBlockParts
  683. } else {
  684. // Create a new proposal block from state/txs from the mempool.
  685. block, blockParts = cs.createProposalBlock()
  686. if block == nil { // on error
  687. return
  688. }
  689. }
  690. // Make proposal
  691. polRound, polBlockID := cs.Votes.POLInfo()
  692. proposal := types.NewProposal(height, round, blockParts.Header(), polRound, polBlockID)
  693. err := cs.privValidator.SignProposal(cs.state.ChainID, proposal)
  694. if err == nil {
  695. // Set fields
  696. /* fields set by setProposal and addBlockPart
  697. cs.Proposal = proposal
  698. cs.ProposalBlock = block
  699. cs.ProposalBlockParts = blockParts
  700. */
  701. // send proposal and block parts on internal msg queue
  702. cs.sendInternalMessage(msgInfo{&ProposalMessage{proposal}, ""})
  703. for i := 0; i < blockParts.Total(); i++ {
  704. part := blockParts.GetPart(i)
  705. cs.sendInternalMessage(msgInfo{&BlockPartMessage{cs.Height, cs.Round, part}, ""})
  706. }
  707. cs.Logger.Info("Signed proposal", "height", height, "round", round, "proposal", proposal)
  708. cs.Logger.Debug(cmn.Fmt("Signed proposal block: %v", block))
  709. } else {
  710. if !cs.replayMode {
  711. cs.Logger.Error("enterPropose: Error signing proposal", "height", height, "round", round, "error", err)
  712. }
  713. }
  714. }
  715. // Returns true if the proposal block is complete &&
  716. // (if POLRound was proposed, we have +2/3 prevotes from there).
  717. func (cs *ConsensusState) isProposalComplete() bool {
  718. if cs.Proposal == nil || cs.ProposalBlock == nil {
  719. return false
  720. }
  721. // we have the proposal. if there's a POLRound,
  722. // make sure we have the prevotes from it too
  723. if cs.Proposal.POLRound < 0 {
  724. return true
  725. } else {
  726. // if this is false the proposer is lying or we haven't received the POL yet
  727. return cs.Votes.Prevotes(cs.Proposal.POLRound).HasTwoThirdsMajority()
  728. }
  729. }
  730. // Create the next block to propose and return it.
  731. // Returns nil block upon error.
  732. // NOTE: keep it side-effect free for clarity.
  733. func (cs *ConsensusState) createProposalBlock() (block *types.Block, blockParts *types.PartSet) {
  734. var commit *types.Commit
  735. if cs.Height == 1 {
  736. // We're creating a proposal for the first block.
  737. // The commit is empty, but not nil.
  738. commit = &types.Commit{}
  739. } else if cs.LastCommit.HasTwoThirdsMajority() {
  740. // Make the commit from LastCommit
  741. commit = cs.LastCommit.MakeCommit()
  742. } else {
  743. // This shouldn't happen.
  744. cs.Logger.Error("enterPropose: Cannot propose anything: No commit for the previous block.")
  745. return
  746. }
  747. // Mempool validated transactions
  748. txs := cs.mempool.Reap(cs.config.MaxBlockSizeTxs)
  749. return types.MakeBlock(cs.Height, cs.state.ChainID, txs, commit,
  750. cs.state.LastBlockID, cs.state.Validators.Hash(), cs.state.AppHash, cs.config.BlockPartSize)
  751. }
  752. // Enter: `timeoutPropose` after entering Propose.
  753. // Enter: proposal block and POL is ready.
  754. // Enter: any +2/3 prevotes for future round.
  755. // Prevote for LockedBlock if we're locked, or ProposalBlock if valid.
  756. // Otherwise vote nil.
  757. func (cs *ConsensusState) enterPrevote(height int, round int) {
  758. if cs.Height != height || round < cs.Round || (cs.Round == round && RoundStepPrevote <= cs.Step) {
  759. cs.Logger.Debug(cmn.Fmt("enterPrevote(%v/%v): Invalid args. Current step: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step))
  760. return
  761. }
  762. defer func() {
  763. // Done enterPrevote:
  764. cs.updateRoundStep(round, RoundStepPrevote)
  765. cs.newStep()
  766. }()
  767. // fire event for how we got here
  768. if cs.isProposalComplete() {
  769. types.FireEventCompleteProposal(cs.evsw, cs.RoundStateEvent())
  770. } else {
  771. // we received +2/3 prevotes for a future round
  772. // TODO: catchup event?
  773. }
  774. cs.Logger.Info(cmn.Fmt("enterPrevote(%v/%v). Current: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step))
  775. // Sign and broadcast vote as necessary
  776. cs.doPrevote(height, round)
  777. // Once `addVote` hits any +2/3 prevotes, we will go to PrevoteWait
  778. // (so we have more time to try and collect +2/3 prevotes for a single block)
  779. }
  780. func (cs *ConsensusState) defaultDoPrevote(height int, round int) {
  781. // If a block is locked, prevote that.
  782. if cs.LockedBlock != nil {
  783. cs.Logger.Info("enterPrevote: Block was locked")
  784. cs.signAddVote(types.VoteTypePrevote, cs.LockedBlock.Hash(), cs.LockedBlockParts.Header())
  785. return
  786. }
  787. // If ProposalBlock is nil, prevote nil.
  788. if cs.ProposalBlock == nil {
  789. cs.Logger.Info("enterPrevote: ProposalBlock is nil")
  790. cs.signAddVote(types.VoteTypePrevote, nil, types.PartSetHeader{})
  791. return
  792. }
  793. // Valdiate proposal block
  794. err := cs.state.ValidateBlock(cs.ProposalBlock)
  795. if err != nil {
  796. // ProposalBlock is invalid, prevote nil.
  797. cs.Logger.Error("enterPrevote: ProposalBlock is invalid", "error", err)
  798. cs.signAddVote(types.VoteTypePrevote, nil, types.PartSetHeader{})
  799. return
  800. }
  801. // Prevote cs.ProposalBlock
  802. // NOTE: the proposal signature is validated when it is received,
  803. // and the proposal block parts are validated as they are received (against the merkle hash in the proposal)
  804. cs.signAddVote(types.VoteTypePrevote, cs.ProposalBlock.Hash(), cs.ProposalBlockParts.Header())
  805. return
  806. }
  807. // Enter: any +2/3 prevotes at next round.
  808. func (cs *ConsensusState) enterPrevoteWait(height int, round int) {
  809. if cs.Height != height || round < cs.Round || (cs.Round == round && RoundStepPrevoteWait <= cs.Step) {
  810. cs.Logger.Debug(cmn.Fmt("enterPrevoteWait(%v/%v): Invalid args. Current step: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step))
  811. return
  812. }
  813. if !cs.Votes.Prevotes(round).HasTwoThirdsAny() {
  814. cmn.PanicSanity(cmn.Fmt("enterPrevoteWait(%v/%v), but Prevotes does not have any +2/3 votes", height, round))
  815. }
  816. cs.Logger.Info(cmn.Fmt("enterPrevoteWait(%v/%v). Current: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step))
  817. defer func() {
  818. // Done enterPrevoteWait:
  819. cs.updateRoundStep(round, RoundStepPrevoteWait)
  820. cs.newStep()
  821. }()
  822. // Wait for some more prevotes; enterPrecommit
  823. cs.scheduleTimeout(cs.config.Prevote(round), height, round, RoundStepPrevoteWait)
  824. }
  825. // Enter: +2/3 precomits for block or nil.
  826. // Enter: `timeoutPrevote` after any +2/3 prevotes.
  827. // Enter: any +2/3 precommits for next round.
  828. // Lock & precommit the ProposalBlock if we have enough prevotes for it (a POL in this round)
  829. // else, unlock an existing lock and precommit nil if +2/3 of prevotes were nil,
  830. // else, precommit nil otherwise.
  831. func (cs *ConsensusState) enterPrecommit(height int, round int) {
  832. if cs.Height != height || round < cs.Round || (cs.Round == round && RoundStepPrecommit <= cs.Step) {
  833. cs.Logger.Debug(cmn.Fmt("enterPrecommit(%v/%v): Invalid args. Current step: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step))
  834. return
  835. }
  836. cs.Logger.Info(cmn.Fmt("enterPrecommit(%v/%v). Current: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step))
  837. defer func() {
  838. // Done enterPrecommit:
  839. cs.updateRoundStep(round, RoundStepPrecommit)
  840. cs.newStep()
  841. }()
  842. blockID, ok := cs.Votes.Prevotes(round).TwoThirdsMajority()
  843. // If we don't have a polka, we must precommit nil
  844. if !ok {
  845. if cs.LockedBlock != nil {
  846. cs.Logger.Info("enterPrecommit: No +2/3 prevotes during enterPrecommit while we're locked. Precommitting nil")
  847. } else {
  848. cs.Logger.Info("enterPrecommit: No +2/3 prevotes during enterPrecommit. Precommitting nil.")
  849. }
  850. cs.signAddVote(types.VoteTypePrecommit, nil, types.PartSetHeader{})
  851. return
  852. }
  853. // At this point +2/3 prevoted for a particular block or nil
  854. types.FireEventPolka(cs.evsw, cs.RoundStateEvent())
  855. // the latest POLRound should be this round
  856. polRound, _ := cs.Votes.POLInfo()
  857. if polRound < round {
  858. cmn.PanicSanity(cmn.Fmt("This POLRound should be %v but got %", round, polRound))
  859. }
  860. // +2/3 prevoted nil. Unlock and precommit nil.
  861. if len(blockID.Hash) == 0 {
  862. if cs.LockedBlock == nil {
  863. cs.Logger.Info("enterPrecommit: +2/3 prevoted for nil.")
  864. } else {
  865. cs.Logger.Info("enterPrecommit: +2/3 prevoted for nil. Unlocking")
  866. cs.LockedRound = 0
  867. cs.LockedBlock = nil
  868. cs.LockedBlockParts = nil
  869. types.FireEventUnlock(cs.evsw, cs.RoundStateEvent())
  870. }
  871. cs.signAddVote(types.VoteTypePrecommit, nil, types.PartSetHeader{})
  872. return
  873. }
  874. // At this point, +2/3 prevoted for a particular block.
  875. // If we're already locked on that block, precommit it, and update the LockedRound
  876. if cs.LockedBlock.HashesTo(blockID.Hash) {
  877. cs.Logger.Info("enterPrecommit: +2/3 prevoted locked block. Relocking")
  878. cs.LockedRound = round
  879. types.FireEventRelock(cs.evsw, cs.RoundStateEvent())
  880. cs.signAddVote(types.VoteTypePrecommit, blockID.Hash, blockID.PartsHeader)
  881. return
  882. }
  883. // If +2/3 prevoted for proposal block, stage and precommit it
  884. if cs.ProposalBlock.HashesTo(blockID.Hash) {
  885. cs.Logger.Info("enterPrecommit: +2/3 prevoted proposal block. Locking", "hash", blockID.Hash)
  886. // Validate the block.
  887. if err := cs.state.ValidateBlock(cs.ProposalBlock); err != nil {
  888. cmn.PanicConsensus(cmn.Fmt("enterPrecommit: +2/3 prevoted for an invalid block: %v", err))
  889. }
  890. cs.LockedRound = round
  891. cs.LockedBlock = cs.ProposalBlock
  892. cs.LockedBlockParts = cs.ProposalBlockParts
  893. types.FireEventLock(cs.evsw, cs.RoundStateEvent())
  894. cs.signAddVote(types.VoteTypePrecommit, blockID.Hash, blockID.PartsHeader)
  895. return
  896. }
  897. // There was a polka in this round for a block we don't have.
  898. // Fetch that block, unlock, and precommit nil.
  899. // The +2/3 prevotes for this round is the POL for our unlock.
  900. // TODO: In the future save the POL prevotes for justification.
  901. cs.LockedRound = 0
  902. cs.LockedBlock = nil
  903. cs.LockedBlockParts = nil
  904. if !cs.ProposalBlockParts.HasHeader(blockID.PartsHeader) {
  905. cs.ProposalBlock = nil
  906. cs.ProposalBlockParts = types.NewPartSetFromHeader(blockID.PartsHeader)
  907. }
  908. types.FireEventUnlock(cs.evsw, cs.RoundStateEvent())
  909. cs.signAddVote(types.VoteTypePrecommit, nil, types.PartSetHeader{})
  910. return
  911. }
  912. // Enter: any +2/3 precommits for next round.
  913. func (cs *ConsensusState) enterPrecommitWait(height int, round int) {
  914. if cs.Height != height || round < cs.Round || (cs.Round == round && RoundStepPrecommitWait <= cs.Step) {
  915. cs.Logger.Debug(cmn.Fmt("enterPrecommitWait(%v/%v): Invalid args. Current step: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step))
  916. return
  917. }
  918. if !cs.Votes.Precommits(round).HasTwoThirdsAny() {
  919. cmn.PanicSanity(cmn.Fmt("enterPrecommitWait(%v/%v), but Precommits does not have any +2/3 votes", height, round))
  920. }
  921. cs.Logger.Info(cmn.Fmt("enterPrecommitWait(%v/%v). Current: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step))
  922. defer func() {
  923. // Done enterPrecommitWait:
  924. cs.updateRoundStep(round, RoundStepPrecommitWait)
  925. cs.newStep()
  926. }()
  927. // Wait for some more precommits; enterNewRound
  928. cs.scheduleTimeout(cs.config.Precommit(round), height, round, RoundStepPrecommitWait)
  929. }
  930. // Enter: +2/3 precommits for block
  931. func (cs *ConsensusState) enterCommit(height int, commitRound int) {
  932. if cs.Height != height || RoundStepCommit <= cs.Step {
  933. cs.Logger.Debug(cmn.Fmt("enterCommit(%v/%v): Invalid args. Current step: %v/%v/%v", height, commitRound, cs.Height, cs.Round, cs.Step))
  934. return
  935. }
  936. cs.Logger.Info(cmn.Fmt("enterCommit(%v/%v). Current: %v/%v/%v", height, commitRound, cs.Height, cs.Round, cs.Step))
  937. defer func() {
  938. // Done enterCommit:
  939. // keep cs.Round the same, commitRound points to the right Precommits set.
  940. cs.updateRoundStep(cs.Round, RoundStepCommit)
  941. cs.CommitRound = commitRound
  942. cs.CommitTime = time.Now()
  943. cs.newStep()
  944. // Maybe finalize immediately.
  945. cs.tryFinalizeCommit(height)
  946. }()
  947. blockID, ok := cs.Votes.Precommits(commitRound).TwoThirdsMajority()
  948. if !ok {
  949. cmn.PanicSanity("RunActionCommit() expects +2/3 precommits")
  950. }
  951. // The Locked* fields no longer matter.
  952. // Move them over to ProposalBlock if they match the commit hash,
  953. // otherwise they'll be cleared in updateToState.
  954. if cs.LockedBlock.HashesTo(blockID.Hash) {
  955. cs.ProposalBlock = cs.LockedBlock
  956. cs.ProposalBlockParts = cs.LockedBlockParts
  957. }
  958. // If we don't have the block being committed, set up to get it.
  959. if !cs.ProposalBlock.HashesTo(blockID.Hash) {
  960. if !cs.ProposalBlockParts.HasHeader(blockID.PartsHeader) {
  961. // We're getting the wrong block.
  962. // Set up ProposalBlockParts and keep waiting.
  963. cs.ProposalBlock = nil
  964. cs.ProposalBlockParts = types.NewPartSetFromHeader(blockID.PartsHeader)
  965. } else {
  966. // We just need to keep waiting.
  967. }
  968. }
  969. }
  970. // If we have the block AND +2/3 commits for it, finalize.
  971. func (cs *ConsensusState) tryFinalizeCommit(height int) {
  972. if cs.Height != height {
  973. cmn.PanicSanity(cmn.Fmt("tryFinalizeCommit() cs.Height: %v vs height: %v", cs.Height, height))
  974. }
  975. blockID, ok := cs.Votes.Precommits(cs.CommitRound).TwoThirdsMajority()
  976. if !ok || len(blockID.Hash) == 0 {
  977. cs.Logger.Error("Attempt to finalize failed. There was no +2/3 majority, or +2/3 was for <nil>.", "height", height)
  978. return
  979. }
  980. if !cs.ProposalBlock.HashesTo(blockID.Hash) {
  981. // TODO: this happens every time if we're not a validator (ugly logs)
  982. // TODO: ^^ wait, why does it matter that we're a validator?
  983. 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)
  984. return
  985. }
  986. // go
  987. cs.finalizeCommit(height)
  988. }
  989. // Increment height and goto RoundStepNewHeight
  990. func (cs *ConsensusState) finalizeCommit(height int) {
  991. if cs.Height != height || cs.Step != RoundStepCommit {
  992. cs.Logger.Debug(cmn.Fmt("finalizeCommit(%v): Invalid args. Current step: %v/%v/%v", height, cs.Height, cs.Round, cs.Step))
  993. return
  994. }
  995. blockID, ok := cs.Votes.Precommits(cs.CommitRound).TwoThirdsMajority()
  996. block, blockParts := cs.ProposalBlock, cs.ProposalBlockParts
  997. if !ok {
  998. cmn.PanicSanity(cmn.Fmt("Cannot finalizeCommit, commit does not have two thirds majority"))
  999. }
  1000. if !blockParts.HasHeader(blockID.PartsHeader) {
  1001. cmn.PanicSanity(cmn.Fmt("Expected ProposalBlockParts header to be commit header"))
  1002. }
  1003. if !block.HashesTo(blockID.Hash) {
  1004. cmn.PanicSanity(cmn.Fmt("Cannot finalizeCommit, ProposalBlock does not hash to commit hash"))
  1005. }
  1006. if err := cs.state.ValidateBlock(block); err != nil {
  1007. cmn.PanicConsensus(cmn.Fmt("+2/3 committed an invalid block: %v", err))
  1008. }
  1009. cs.Logger.Info(cmn.Fmt("Finalizing commit of block with %d txs", block.NumTxs),
  1010. "height", block.Height, "hash", block.Hash(), "root", block.AppHash)
  1011. cs.Logger.Info(cmn.Fmt("%v", block))
  1012. fail.Fail() // XXX
  1013. // Save to blockStore.
  1014. if cs.blockStore.Height() < block.Height {
  1015. // NOTE: the seenCommit is local justification to commit this block,
  1016. // but may differ from the LastCommit included in the next block
  1017. precommits := cs.Votes.Precommits(cs.CommitRound)
  1018. seenCommit := precommits.MakeCommit()
  1019. cs.blockStore.SaveBlock(block, blockParts, seenCommit)
  1020. } else {
  1021. // Happens during replay if we already saved the block but didn't commit
  1022. cs.Logger.Info("Calling finalizeCommit on already stored block", "height", block.Height)
  1023. }
  1024. fail.Fail() // XXX
  1025. // Finish writing to the WAL for this height.
  1026. // NOTE: If we fail before writing this, we'll never write it,
  1027. // and just recover by running ApplyBlock in the Handshake.
  1028. // If we moved it before persisting the block, we'd have to allow
  1029. // WAL replay for blocks with an #ENDHEIGHT
  1030. // As is, ConsensusState should not be started again
  1031. // until we successfully call ApplyBlock (ie. here or in Handshake after restart)
  1032. if cs.wal != nil {
  1033. cs.wal.writeEndHeight(height)
  1034. }
  1035. fail.Fail() // XXX
  1036. // Create a copy of the state for staging
  1037. // and an event cache for txs
  1038. stateCopy := cs.state.Copy()
  1039. eventCache := types.NewEventCache(cs.evsw)
  1040. // Execute and commit the block, update and save the state, and update the mempool.
  1041. // All calls to the proxyAppConn come here.
  1042. // NOTE: the block.AppHash wont reflect these txs until the next block
  1043. err := stateCopy.ApplyBlock(eventCache, cs.proxyAppConn, block, blockParts.Header(), cs.mempool)
  1044. if err != nil {
  1045. cs.Logger.Error("Error on ApplyBlock. Did the application crash? Please restart tendermint", "error", err)
  1046. return
  1047. }
  1048. fail.Fail() // XXX
  1049. // Fire event for new block.
  1050. // NOTE: If we fail before firing, these events will never fire
  1051. //
  1052. // TODO: Either
  1053. // * Fire before persisting state, in ApplyBlock
  1054. // * Fire on start up if we haven't written any new WAL msgs
  1055. // Both options mean we may fire more than once. Is that fine ?
  1056. types.FireEventNewBlock(cs.evsw, types.EventDataNewBlock{block})
  1057. types.FireEventNewBlockHeader(cs.evsw, types.EventDataNewBlockHeader{block.Header})
  1058. eventCache.Flush()
  1059. fail.Fail() // XXX
  1060. // NewHeightStep!
  1061. cs.updateToState(stateCopy)
  1062. fail.Fail() // XXX
  1063. // cs.StartTime is already set.
  1064. // Schedule Round0 to start soon.
  1065. cs.scheduleRound0(&cs.RoundState)
  1066. // By here,
  1067. // * cs.Height has been increment to height+1
  1068. // * cs.Step is now RoundStepNewHeight
  1069. // * cs.StartTime is set to when we will start round0.
  1070. return
  1071. }
  1072. //-----------------------------------------------------------------------------
  1073. func (cs *ConsensusState) defaultSetProposal(proposal *types.Proposal) error {
  1074. // Already have one
  1075. // TODO: possibly catch double proposals
  1076. if cs.Proposal != nil {
  1077. return nil
  1078. }
  1079. // Does not apply
  1080. if proposal.Height != cs.Height || proposal.Round != cs.Round {
  1081. return nil
  1082. }
  1083. // We don't care about the proposal if we're already in RoundStepCommit.
  1084. if RoundStepCommit <= cs.Step {
  1085. return nil
  1086. }
  1087. // Verify POLRound, which must be -1 or between 0 and proposal.Round exclusive.
  1088. if proposal.POLRound != -1 &&
  1089. (proposal.POLRound < 0 || proposal.Round <= proposal.POLRound) {
  1090. return ErrInvalidProposalPOLRound
  1091. }
  1092. // Verify signature
  1093. if !cs.Validators.GetProposer().PubKey.VerifyBytes(types.SignBytes(cs.state.ChainID, proposal), proposal.Signature) {
  1094. return ErrInvalidProposalSignature
  1095. }
  1096. cs.Proposal = proposal
  1097. cs.ProposalBlockParts = types.NewPartSetFromHeader(proposal.BlockPartsHeader)
  1098. return nil
  1099. }
  1100. // NOTE: block is not necessarily valid.
  1101. // Asynchronously triggers either enterPrevote (before we timeout of propose) or tryFinalizeCommit, once we have the full block.
  1102. func (cs *ConsensusState) addProposalBlockPart(height int, part *types.Part, verify bool) (added bool, err error) {
  1103. // Blocks might be reused, so round mismatch is OK
  1104. if cs.Height != height {
  1105. return false, nil
  1106. }
  1107. // We're not expecting a block part.
  1108. if cs.ProposalBlockParts == nil {
  1109. return false, nil // TODO: bad peer? Return error?
  1110. }
  1111. added, err = cs.ProposalBlockParts.AddPart(part, verify)
  1112. if err != nil {
  1113. return added, err
  1114. }
  1115. if added && cs.ProposalBlockParts.IsComplete() {
  1116. // Added and completed!
  1117. var n int
  1118. var err error
  1119. cs.ProposalBlock = wire.ReadBinary(&types.Block{}, cs.ProposalBlockParts.GetReader(), types.MaxBlockSize, &n, &err).(*types.Block)
  1120. // NOTE: it's possible to receive complete proposal blocks for future rounds without having the proposal
  1121. cs.Logger.Info("Received complete proposal block", "height", cs.ProposalBlock.Height, "hash", cs.ProposalBlock.Hash())
  1122. if cs.Step == RoundStepPropose && cs.isProposalComplete() {
  1123. // Move onto the next step
  1124. cs.enterPrevote(height, cs.Round)
  1125. } else if cs.Step == RoundStepCommit {
  1126. // If we're waiting on the proposal block...
  1127. cs.tryFinalizeCommit(height)
  1128. }
  1129. return true, err
  1130. }
  1131. return added, nil
  1132. }
  1133. // Attempt to add the vote. if its a duplicate signature, dupeout the validator
  1134. func (cs *ConsensusState) tryAddVote(vote *types.Vote, peerKey string) error {
  1135. _, err := cs.addVote(vote, peerKey)
  1136. if err != nil {
  1137. // If the vote height is off, we'll just ignore it,
  1138. // But if it's a conflicting sig, broadcast evidence tx for slashing.
  1139. // If it's otherwise invalid, punish peer.
  1140. if err == ErrVoteHeightMismatch {
  1141. return err
  1142. } else if _, ok := err.(*types.ErrVoteConflictingVotes); ok {
  1143. if peerKey == "" {
  1144. cs.Logger.Error("Found conflicting vote from ourselves. Did you unsafe_reset a validator?", "height", vote.Height, "round", vote.Round, "type", vote.Type)
  1145. return err
  1146. }
  1147. cs.Logger.Error("Found conflicting vote. Publish evidence (TODO)")
  1148. /* TODO
  1149. evidenceTx := &types.DupeoutTx{
  1150. Address: address,
  1151. VoteA: *errDupe.VoteA,
  1152. VoteB: *errDupe.VoteB,
  1153. }
  1154. cs.mempool.BroadcastTx(struct{???}{evidenceTx}) // shouldn't need to check returned err
  1155. */
  1156. return err
  1157. } else {
  1158. // Probably an invalid signature. Bad peer.
  1159. cs.Logger.Error("Error attempting to add vote", "error", err)
  1160. return ErrAddingVote
  1161. }
  1162. }
  1163. return nil
  1164. }
  1165. //-----------------------------------------------------------------------------
  1166. func (cs *ConsensusState) addVote(vote *types.Vote, peerKey string) (added bool, err error) {
  1167. cs.Logger.Debug("addVote", "voteHeight", vote.Height, "voteType", vote.Type, "csHeight", cs.Height)
  1168. // A precommit for the previous height?
  1169. // These come in while we wait timeoutCommit
  1170. if vote.Height+1 == cs.Height {
  1171. if !(cs.Step == RoundStepNewHeight && vote.Type == types.VoteTypePrecommit) {
  1172. // TODO: give the reason ..
  1173. // fmt.Errorf("tryAddVote: Wrong height, not a LastCommit straggler commit.")
  1174. return added, ErrVoteHeightMismatch
  1175. }
  1176. added, err = cs.LastCommit.AddVote(vote)
  1177. if added {
  1178. cs.Logger.Info(cmn.Fmt("Added to lastPrecommits: %v", cs.LastCommit.StringShort()))
  1179. types.FireEventVote(cs.evsw, types.EventDataVote{vote})
  1180. // if we can skip timeoutCommit and have all the votes now,
  1181. if cs.config.SkipTimeoutCommit && cs.LastCommit.HasAll() {
  1182. // go straight to new round (skip timeout commit)
  1183. // cs.scheduleTimeout(time.Duration(0), cs.Height, 0, RoundStepNewHeight)
  1184. cs.enterNewRound(cs.Height, 0)
  1185. }
  1186. }
  1187. return
  1188. }
  1189. // A prevote/precommit for this height?
  1190. if vote.Height == cs.Height {
  1191. height := cs.Height
  1192. added, err = cs.Votes.AddVote(vote, peerKey)
  1193. if added {
  1194. types.FireEventVote(cs.evsw, types.EventDataVote{vote})
  1195. switch vote.Type {
  1196. case types.VoteTypePrevote:
  1197. prevotes := cs.Votes.Prevotes(vote.Round)
  1198. cs.Logger.Info("Added to prevote", "vote", vote, "prevotes", prevotes.StringShort())
  1199. // First, unlock if prevotes is a valid POL.
  1200. // >> lockRound < POLRound <= unlockOrChangeLockRound (see spec)
  1201. // NOTE: If (lockRound < POLRound) but !(POLRound <= unlockOrChangeLockRound),
  1202. // we'll still enterNewRound(H,vote.R) and enterPrecommit(H,vote.R) to process it
  1203. // there.
  1204. if (cs.LockedBlock != nil) && (cs.LockedRound < vote.Round) && (vote.Round <= cs.Round) {
  1205. blockID, ok := prevotes.TwoThirdsMajority()
  1206. if ok && !cs.LockedBlock.HashesTo(blockID.Hash) {
  1207. cs.Logger.Info("Unlocking because of POL.", "lockedRound", cs.LockedRound, "POLRound", vote.Round)
  1208. cs.LockedRound = 0
  1209. cs.LockedBlock = nil
  1210. cs.LockedBlockParts = nil
  1211. types.FireEventUnlock(cs.evsw, cs.RoundStateEvent())
  1212. }
  1213. }
  1214. if cs.Round <= vote.Round && prevotes.HasTwoThirdsAny() {
  1215. // Round-skip over to PrevoteWait or goto Precommit.
  1216. cs.enterNewRound(height, vote.Round) // if the vote is ahead of us
  1217. if prevotes.HasTwoThirdsMajority() {
  1218. cs.enterPrecommit(height, vote.Round)
  1219. } else {
  1220. cs.enterPrevote(height, vote.Round) // if the vote is ahead of us
  1221. cs.enterPrevoteWait(height, vote.Round)
  1222. }
  1223. } else if cs.Proposal != nil && 0 <= cs.Proposal.POLRound && cs.Proposal.POLRound == vote.Round {
  1224. // If the proposal is now complete, enter prevote of cs.Round.
  1225. if cs.isProposalComplete() {
  1226. cs.enterPrevote(height, cs.Round)
  1227. }
  1228. }
  1229. case types.VoteTypePrecommit:
  1230. precommits := cs.Votes.Precommits(vote.Round)
  1231. cs.Logger.Info("Added to precommit", "vote", vote, "precommits", precommits.StringShort())
  1232. blockID, ok := precommits.TwoThirdsMajority()
  1233. if ok {
  1234. if len(blockID.Hash) == 0 {
  1235. cs.enterNewRound(height, vote.Round+1)
  1236. } else {
  1237. cs.enterNewRound(height, vote.Round)
  1238. cs.enterPrecommit(height, vote.Round)
  1239. cs.enterCommit(height, vote.Round)
  1240. if cs.config.SkipTimeoutCommit && precommits.HasAll() {
  1241. // if we have all the votes now,
  1242. // go straight to new round (skip timeout commit)
  1243. // cs.scheduleTimeout(time.Duration(0), cs.Height, 0, RoundStepNewHeight)
  1244. cs.enterNewRound(cs.Height, 0)
  1245. }
  1246. }
  1247. } else if cs.Round <= vote.Round && precommits.HasTwoThirdsAny() {
  1248. cs.enterNewRound(height, vote.Round)
  1249. cs.enterPrecommit(height, vote.Round)
  1250. cs.enterPrecommitWait(height, vote.Round)
  1251. }
  1252. default:
  1253. cmn.PanicSanity(cmn.Fmt("Unexpected vote type %X", vote.Type)) // Should not happen.
  1254. }
  1255. }
  1256. // Either duplicate, or error upon cs.Votes.AddByIndex()
  1257. return
  1258. } else {
  1259. err = ErrVoteHeightMismatch
  1260. }
  1261. // Height mismatch, bad peer?
  1262. cs.Logger.Info("Vote ignored and not added", "voteHeight", vote.Height, "csHeight", cs.Height, "err", err)
  1263. return
  1264. }
  1265. func (cs *ConsensusState) signVote(type_ byte, hash []byte, header types.PartSetHeader) (*types.Vote, error) {
  1266. addr := cs.privValidator.GetAddress()
  1267. valIndex, _ := cs.Validators.GetByAddress(addr)
  1268. vote := &types.Vote{
  1269. ValidatorAddress: addr,
  1270. ValidatorIndex: valIndex,
  1271. Height: cs.Height,
  1272. Round: cs.Round,
  1273. Type: type_,
  1274. BlockID: types.BlockID{hash, header},
  1275. }
  1276. err := cs.privValidator.SignVote(cs.state.ChainID, vote)
  1277. return vote, err
  1278. }
  1279. // sign the vote and publish on internalMsgQueue
  1280. func (cs *ConsensusState) signAddVote(type_ byte, hash []byte, header types.PartSetHeader) *types.Vote {
  1281. // if we don't have a key or we're not in the validator set, do nothing
  1282. if cs.privValidator == nil || !cs.Validators.HasAddress(cs.privValidator.GetAddress()) {
  1283. return nil
  1284. }
  1285. vote, err := cs.signVote(type_, hash, header)
  1286. if err == nil {
  1287. cs.sendInternalMessage(msgInfo{&VoteMessage{vote}, ""})
  1288. cs.Logger.Info("Signed and pushed vote", "height", cs.Height, "round", cs.Round, "vote", vote, "error", err)
  1289. return vote
  1290. } else {
  1291. //if !cs.replayMode {
  1292. cs.Logger.Error("Error signing vote", "height", cs.Height, "round", cs.Round, "vote", vote, "error", err)
  1293. //}
  1294. return nil
  1295. }
  1296. }
  1297. //---------------------------------------------------------
  1298. func CompareHRS(h1, r1 int, s1 RoundStepType, h2, r2 int, s2 RoundStepType) int {
  1299. if h1 < h2 {
  1300. return -1
  1301. } else if h1 > h2 {
  1302. return 1
  1303. }
  1304. if r1 < r2 {
  1305. return -1
  1306. } else if r1 > r2 {
  1307. return 1
  1308. }
  1309. if s1 < s2 {
  1310. return -1
  1311. } else if s1 > s2 {
  1312. return 1
  1313. }
  1314. return 0
  1315. }