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.

1523 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
9 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
9 years ago
8 years ago
10 years ago
8 years ago
10 years ago
8 years ago
9 years ago
9 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
9 years ago
9 years ago
9 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
9 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
9 years ago
9 years ago
8 years ago
9 years ago
8 years ago
9 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
9 years ago
9 years ago
10 years ago
10 years ago
10 years ago
9 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
9 years ago
8 years ago
9 years ago
9 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. cs.Logger.Info("This node is not a validator")
  668. return
  669. }
  670. if !bytes.Equal(cs.Validators.GetProposer().Address, cs.privValidator.GetAddress()) {
  671. cs.Logger.Info("enterPropose: Not our turn to propose", "proposer", cs.Validators.GetProposer().Address, "privValidator", cs.privValidator)
  672. if cs.Validators.HasAddress(cs.privValidator.GetAddress()) {
  673. cs.Logger.Info("This node is a validator")
  674. } else {
  675. cs.Logger.Info("This node is not a validator")
  676. }
  677. } else {
  678. cs.Logger.Info("enterPropose: Our turn to propose", "proposer", cs.Validators.GetProposer().Address, "privValidator", cs.privValidator)
  679. cs.Logger.Info("This node is a validator")
  680. cs.decideProposal(height, round)
  681. }
  682. }
  683. func (cs *ConsensusState) defaultDecideProposal(height, round int) {
  684. var block *types.Block
  685. var blockParts *types.PartSet
  686. // Decide on block
  687. if cs.LockedBlock != nil {
  688. // If we're locked onto a block, just choose that.
  689. block, blockParts = cs.LockedBlock, cs.LockedBlockParts
  690. } else {
  691. // Create a new proposal block from state/txs from the mempool.
  692. block, blockParts = cs.createProposalBlock()
  693. if block == nil { // on error
  694. return
  695. }
  696. }
  697. // Make proposal
  698. polRound, polBlockID := cs.Votes.POLInfo()
  699. proposal := types.NewProposal(height, round, blockParts.Header(), polRound, polBlockID)
  700. err := cs.privValidator.SignProposal(cs.state.ChainID, proposal)
  701. if err == nil {
  702. // Set fields
  703. /* fields set by setProposal and addBlockPart
  704. cs.Proposal = proposal
  705. cs.ProposalBlock = block
  706. cs.ProposalBlockParts = blockParts
  707. */
  708. // send proposal and block parts on internal msg queue
  709. cs.sendInternalMessage(msgInfo{&ProposalMessage{proposal}, ""})
  710. for i := 0; i < blockParts.Total(); i++ {
  711. part := blockParts.GetPart(i)
  712. cs.sendInternalMessage(msgInfo{&BlockPartMessage{cs.Height, cs.Round, part}, ""})
  713. }
  714. cs.Logger.Info("Signed proposal", "height", height, "round", round, "proposal", proposal)
  715. cs.Logger.Debug(cmn.Fmt("Signed proposal block: %v", block))
  716. } else {
  717. if !cs.replayMode {
  718. cs.Logger.Error("enterPropose: Error signing proposal", "height", height, "round", round, "error", err)
  719. }
  720. }
  721. }
  722. // Returns true if the proposal block is complete &&
  723. // (if POLRound was proposed, we have +2/3 prevotes from there).
  724. func (cs *ConsensusState) isProposalComplete() bool {
  725. if cs.Proposal == nil || cs.ProposalBlock == nil {
  726. return false
  727. }
  728. // we have the proposal. if there's a POLRound,
  729. // make sure we have the prevotes from it too
  730. if cs.Proposal.POLRound < 0 {
  731. return true
  732. } else {
  733. // if this is false the proposer is lying or we haven't received the POL yet
  734. return cs.Votes.Prevotes(cs.Proposal.POLRound).HasTwoThirdsMajority()
  735. }
  736. }
  737. // Create the next block to propose and return it.
  738. // Returns nil block upon error.
  739. // NOTE: keep it side-effect free for clarity.
  740. func (cs *ConsensusState) createProposalBlock() (block *types.Block, blockParts *types.PartSet) {
  741. var commit *types.Commit
  742. if cs.Height == 1 {
  743. // We're creating a proposal for the first block.
  744. // The commit is empty, but not nil.
  745. commit = &types.Commit{}
  746. } else if cs.LastCommit.HasTwoThirdsMajority() {
  747. // Make the commit from LastCommit
  748. commit = cs.LastCommit.MakeCommit()
  749. } else {
  750. // This shouldn't happen.
  751. cs.Logger.Error("enterPropose: Cannot propose anything: No commit for the previous block.")
  752. return
  753. }
  754. // Mempool validated transactions
  755. txs := cs.mempool.Reap(cs.config.MaxBlockSizeTxs)
  756. return types.MakeBlock(cs.Height, cs.state.ChainID, txs, commit,
  757. cs.state.LastBlockID, cs.state.Validators.Hash(), cs.state.AppHash, cs.config.BlockPartSize)
  758. }
  759. // Enter: `timeoutPropose` after entering Propose.
  760. // Enter: proposal block and POL is ready.
  761. // Enter: any +2/3 prevotes for future round.
  762. // Prevote for LockedBlock if we're locked, or ProposalBlock if valid.
  763. // Otherwise vote nil.
  764. func (cs *ConsensusState) enterPrevote(height int, round int) {
  765. if cs.Height != height || round < cs.Round || (cs.Round == round && RoundStepPrevote <= cs.Step) {
  766. cs.Logger.Debug(cmn.Fmt("enterPrevote(%v/%v): Invalid args. Current step: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step))
  767. return
  768. }
  769. defer func() {
  770. // Done enterPrevote:
  771. cs.updateRoundStep(round, RoundStepPrevote)
  772. cs.newStep()
  773. }()
  774. // fire event for how we got here
  775. if cs.isProposalComplete() {
  776. types.FireEventCompleteProposal(cs.evsw, cs.RoundStateEvent())
  777. } else {
  778. // we received +2/3 prevotes for a future round
  779. // TODO: catchup event?
  780. }
  781. cs.Logger.Info(cmn.Fmt("enterPrevote(%v/%v). Current: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step))
  782. // Sign and broadcast vote as necessary
  783. cs.doPrevote(height, round)
  784. // Once `addVote` hits any +2/3 prevotes, we will go to PrevoteWait
  785. // (so we have more time to try and collect +2/3 prevotes for a single block)
  786. }
  787. func (cs *ConsensusState) defaultDoPrevote(height int, round int) {
  788. // If a block is locked, prevote that.
  789. if cs.LockedBlock != nil {
  790. cs.Logger.Info("enterPrevote: Block was locked")
  791. cs.signAddVote(types.VoteTypePrevote, cs.LockedBlock.Hash(), cs.LockedBlockParts.Header())
  792. return
  793. }
  794. // If ProposalBlock is nil, prevote nil.
  795. if cs.ProposalBlock == nil {
  796. cs.Logger.Info("enterPrevote: ProposalBlock is nil")
  797. cs.signAddVote(types.VoteTypePrevote, nil, types.PartSetHeader{})
  798. return
  799. }
  800. // Valdiate proposal block
  801. err := cs.state.ValidateBlock(cs.ProposalBlock)
  802. if err != nil {
  803. // ProposalBlock is invalid, prevote nil.
  804. cs.Logger.Error("enterPrevote: ProposalBlock is invalid", "error", err)
  805. cs.signAddVote(types.VoteTypePrevote, nil, types.PartSetHeader{})
  806. return
  807. }
  808. // Prevote cs.ProposalBlock
  809. // NOTE: the proposal signature is validated when it is received,
  810. // and the proposal block parts are validated as they are received (against the merkle hash in the proposal)
  811. cs.signAddVote(types.VoteTypePrevote, cs.ProposalBlock.Hash(), cs.ProposalBlockParts.Header())
  812. return
  813. }
  814. // Enter: any +2/3 prevotes at next round.
  815. func (cs *ConsensusState) enterPrevoteWait(height int, round int) {
  816. if cs.Height != height || round < cs.Round || (cs.Round == round && RoundStepPrevoteWait <= cs.Step) {
  817. cs.Logger.Debug(cmn.Fmt("enterPrevoteWait(%v/%v): Invalid args. Current step: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step))
  818. return
  819. }
  820. if !cs.Votes.Prevotes(round).HasTwoThirdsAny() {
  821. cmn.PanicSanity(cmn.Fmt("enterPrevoteWait(%v/%v), but Prevotes does not have any +2/3 votes", height, round))
  822. }
  823. cs.Logger.Info(cmn.Fmt("enterPrevoteWait(%v/%v). Current: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step))
  824. defer func() {
  825. // Done enterPrevoteWait:
  826. cs.updateRoundStep(round, RoundStepPrevoteWait)
  827. cs.newStep()
  828. }()
  829. // Wait for some more prevotes; enterPrecommit
  830. cs.scheduleTimeout(cs.config.Prevote(round), height, round, RoundStepPrevoteWait)
  831. }
  832. // Enter: +2/3 precomits for block or nil.
  833. // Enter: `timeoutPrevote` after any +2/3 prevotes.
  834. // Enter: any +2/3 precommits for next round.
  835. // Lock & precommit the ProposalBlock if we have enough prevotes for it (a POL in this round)
  836. // else, unlock an existing lock and precommit nil if +2/3 of prevotes were nil,
  837. // else, precommit nil otherwise.
  838. func (cs *ConsensusState) enterPrecommit(height int, round int) {
  839. if cs.Height != height || round < cs.Round || (cs.Round == round && RoundStepPrecommit <= cs.Step) {
  840. cs.Logger.Debug(cmn.Fmt("enterPrecommit(%v/%v): Invalid args. Current step: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step))
  841. return
  842. }
  843. cs.Logger.Info(cmn.Fmt("enterPrecommit(%v/%v). Current: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step))
  844. defer func() {
  845. // Done enterPrecommit:
  846. cs.updateRoundStep(round, RoundStepPrecommit)
  847. cs.newStep()
  848. }()
  849. blockID, ok := cs.Votes.Prevotes(round).TwoThirdsMajority()
  850. // If we don't have a polka, we must precommit nil
  851. if !ok {
  852. if cs.LockedBlock != nil {
  853. cs.Logger.Info("enterPrecommit: No +2/3 prevotes during enterPrecommit while we're locked. Precommitting nil")
  854. } else {
  855. cs.Logger.Info("enterPrecommit: No +2/3 prevotes during enterPrecommit. Precommitting nil.")
  856. }
  857. cs.signAddVote(types.VoteTypePrecommit, nil, types.PartSetHeader{})
  858. return
  859. }
  860. // At this point +2/3 prevoted for a particular block or nil
  861. types.FireEventPolka(cs.evsw, cs.RoundStateEvent())
  862. // the latest POLRound should be this round
  863. polRound, _ := cs.Votes.POLInfo()
  864. if polRound < round {
  865. cmn.PanicSanity(cmn.Fmt("This POLRound should be %v but got %", round, polRound))
  866. }
  867. // +2/3 prevoted nil. Unlock and precommit nil.
  868. if len(blockID.Hash) == 0 {
  869. if cs.LockedBlock == nil {
  870. cs.Logger.Info("enterPrecommit: +2/3 prevoted for nil.")
  871. } else {
  872. cs.Logger.Info("enterPrecommit: +2/3 prevoted for nil. Unlocking")
  873. cs.LockedRound = 0
  874. cs.LockedBlock = nil
  875. cs.LockedBlockParts = nil
  876. types.FireEventUnlock(cs.evsw, cs.RoundStateEvent())
  877. }
  878. cs.signAddVote(types.VoteTypePrecommit, nil, types.PartSetHeader{})
  879. return
  880. }
  881. // At this point, +2/3 prevoted for a particular block.
  882. // If we're already locked on that block, precommit it, and update the LockedRound
  883. if cs.LockedBlock.HashesTo(blockID.Hash) {
  884. cs.Logger.Info("enterPrecommit: +2/3 prevoted locked block. Relocking")
  885. cs.LockedRound = round
  886. types.FireEventRelock(cs.evsw, cs.RoundStateEvent())
  887. cs.signAddVote(types.VoteTypePrecommit, blockID.Hash, blockID.PartsHeader)
  888. return
  889. }
  890. // If +2/3 prevoted for proposal block, stage and precommit it
  891. if cs.ProposalBlock.HashesTo(blockID.Hash) {
  892. cs.Logger.Info("enterPrecommit: +2/3 prevoted proposal block. Locking", "hash", blockID.Hash)
  893. // Validate the block.
  894. if err := cs.state.ValidateBlock(cs.ProposalBlock); err != nil {
  895. cmn.PanicConsensus(cmn.Fmt("enterPrecommit: +2/3 prevoted for an invalid block: %v", err))
  896. }
  897. cs.LockedRound = round
  898. cs.LockedBlock = cs.ProposalBlock
  899. cs.LockedBlockParts = cs.ProposalBlockParts
  900. types.FireEventLock(cs.evsw, cs.RoundStateEvent())
  901. cs.signAddVote(types.VoteTypePrecommit, blockID.Hash, blockID.PartsHeader)
  902. return
  903. }
  904. // There was a polka in this round for a block we don't have.
  905. // Fetch that block, unlock, and precommit nil.
  906. // The +2/3 prevotes for this round is the POL for our unlock.
  907. // TODO: In the future save the POL prevotes for justification.
  908. cs.LockedRound = 0
  909. cs.LockedBlock = nil
  910. cs.LockedBlockParts = nil
  911. if !cs.ProposalBlockParts.HasHeader(blockID.PartsHeader) {
  912. cs.ProposalBlock = nil
  913. cs.ProposalBlockParts = types.NewPartSetFromHeader(blockID.PartsHeader)
  914. }
  915. types.FireEventUnlock(cs.evsw, cs.RoundStateEvent())
  916. cs.signAddVote(types.VoteTypePrecommit, nil, types.PartSetHeader{})
  917. return
  918. }
  919. // Enter: any +2/3 precommits for next round.
  920. func (cs *ConsensusState) enterPrecommitWait(height int, round int) {
  921. if cs.Height != height || round < cs.Round || (cs.Round == round && RoundStepPrecommitWait <= cs.Step) {
  922. cs.Logger.Debug(cmn.Fmt("enterPrecommitWait(%v/%v): Invalid args. Current step: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step))
  923. return
  924. }
  925. if !cs.Votes.Precommits(round).HasTwoThirdsAny() {
  926. cmn.PanicSanity(cmn.Fmt("enterPrecommitWait(%v/%v), but Precommits does not have any +2/3 votes", height, round))
  927. }
  928. cs.Logger.Info(cmn.Fmt("enterPrecommitWait(%v/%v). Current: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step))
  929. defer func() {
  930. // Done enterPrecommitWait:
  931. cs.updateRoundStep(round, RoundStepPrecommitWait)
  932. cs.newStep()
  933. }()
  934. // Wait for some more precommits; enterNewRound
  935. cs.scheduleTimeout(cs.config.Precommit(round), height, round, RoundStepPrecommitWait)
  936. }
  937. // Enter: +2/3 precommits for block
  938. func (cs *ConsensusState) enterCommit(height int, commitRound int) {
  939. if cs.Height != height || RoundStepCommit <= cs.Step {
  940. cs.Logger.Debug(cmn.Fmt("enterCommit(%v/%v): Invalid args. Current step: %v/%v/%v", height, commitRound, cs.Height, cs.Round, cs.Step))
  941. return
  942. }
  943. cs.Logger.Info(cmn.Fmt("enterCommit(%v/%v). Current: %v/%v/%v", height, commitRound, cs.Height, cs.Round, cs.Step))
  944. defer func() {
  945. // Done enterCommit:
  946. // keep cs.Round the same, commitRound points to the right Precommits set.
  947. cs.updateRoundStep(cs.Round, RoundStepCommit)
  948. cs.CommitRound = commitRound
  949. cs.CommitTime = time.Now()
  950. cs.newStep()
  951. // Maybe finalize immediately.
  952. cs.tryFinalizeCommit(height)
  953. }()
  954. blockID, ok := cs.Votes.Precommits(commitRound).TwoThirdsMajority()
  955. if !ok {
  956. cmn.PanicSanity("RunActionCommit() expects +2/3 precommits")
  957. }
  958. // The Locked* fields no longer matter.
  959. // Move them over to ProposalBlock if they match the commit hash,
  960. // otherwise they'll be cleared in updateToState.
  961. if cs.LockedBlock.HashesTo(blockID.Hash) {
  962. cs.ProposalBlock = cs.LockedBlock
  963. cs.ProposalBlockParts = cs.LockedBlockParts
  964. }
  965. // If we don't have the block being committed, set up to get it.
  966. if !cs.ProposalBlock.HashesTo(blockID.Hash) {
  967. if !cs.ProposalBlockParts.HasHeader(blockID.PartsHeader) {
  968. // We're getting the wrong block.
  969. // Set up ProposalBlockParts and keep waiting.
  970. cs.ProposalBlock = nil
  971. cs.ProposalBlockParts = types.NewPartSetFromHeader(blockID.PartsHeader)
  972. } else {
  973. // We just need to keep waiting.
  974. }
  975. }
  976. }
  977. // If we have the block AND +2/3 commits for it, finalize.
  978. func (cs *ConsensusState) tryFinalizeCommit(height int) {
  979. if cs.Height != height {
  980. cmn.PanicSanity(cmn.Fmt("tryFinalizeCommit() cs.Height: %v vs height: %v", cs.Height, height))
  981. }
  982. blockID, ok := cs.Votes.Precommits(cs.CommitRound).TwoThirdsMajority()
  983. if !ok || len(blockID.Hash) == 0 {
  984. cs.Logger.Error("Attempt to finalize failed. There was no +2/3 majority, or +2/3 was for <nil>.", "height", height)
  985. return
  986. }
  987. if !cs.ProposalBlock.HashesTo(blockID.Hash) {
  988. // TODO: this happens every time if we're not a validator (ugly logs)
  989. // TODO: ^^ wait, why does it matter that we're a validator?
  990. 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)
  991. return
  992. }
  993. // go
  994. cs.finalizeCommit(height)
  995. }
  996. // Increment height and goto RoundStepNewHeight
  997. func (cs *ConsensusState) finalizeCommit(height int) {
  998. if cs.Height != height || cs.Step != RoundStepCommit {
  999. cs.Logger.Debug(cmn.Fmt("finalizeCommit(%v): Invalid args. Current step: %v/%v/%v", height, cs.Height, cs.Round, cs.Step))
  1000. return
  1001. }
  1002. blockID, ok := cs.Votes.Precommits(cs.CommitRound).TwoThirdsMajority()
  1003. block, blockParts := cs.ProposalBlock, cs.ProposalBlockParts
  1004. if !ok {
  1005. cmn.PanicSanity(cmn.Fmt("Cannot finalizeCommit, commit does not have two thirds majority"))
  1006. }
  1007. if !blockParts.HasHeader(blockID.PartsHeader) {
  1008. cmn.PanicSanity(cmn.Fmt("Expected ProposalBlockParts header to be commit header"))
  1009. }
  1010. if !block.HashesTo(blockID.Hash) {
  1011. cmn.PanicSanity(cmn.Fmt("Cannot finalizeCommit, ProposalBlock does not hash to commit hash"))
  1012. }
  1013. if err := cs.state.ValidateBlock(block); err != nil {
  1014. cmn.PanicConsensus(cmn.Fmt("+2/3 committed an invalid block: %v", err))
  1015. }
  1016. cs.Logger.Info(cmn.Fmt("Finalizing commit of block with %d txs", block.NumTxs),
  1017. "height", block.Height, "hash", block.Hash(), "root", block.AppHash)
  1018. cs.Logger.Info(cmn.Fmt("%v", block))
  1019. fail.Fail() // XXX
  1020. // Save to blockStore.
  1021. if cs.blockStore.Height() < block.Height {
  1022. // NOTE: the seenCommit is local justification to commit this block,
  1023. // but may differ from the LastCommit included in the next block
  1024. precommits := cs.Votes.Precommits(cs.CommitRound)
  1025. seenCommit := precommits.MakeCommit()
  1026. cs.blockStore.SaveBlock(block, blockParts, seenCommit)
  1027. } else {
  1028. // Happens during replay if we already saved the block but didn't commit
  1029. cs.Logger.Info("Calling finalizeCommit on already stored block", "height", block.Height)
  1030. }
  1031. fail.Fail() // XXX
  1032. // Finish writing to the WAL for this height.
  1033. // NOTE: If we fail before writing this, we'll never write it,
  1034. // and just recover by running ApplyBlock in the Handshake.
  1035. // If we moved it before persisting the block, we'd have to allow
  1036. // WAL replay for blocks with an #ENDHEIGHT
  1037. // As is, ConsensusState should not be started again
  1038. // until we successfully call ApplyBlock (ie. here or in Handshake after restart)
  1039. if cs.wal != nil {
  1040. cs.wal.writeEndHeight(height)
  1041. }
  1042. fail.Fail() // XXX
  1043. // Create a copy of the state for staging
  1044. // and an event cache for txs
  1045. stateCopy := cs.state.Copy()
  1046. eventCache := types.NewEventCache(cs.evsw)
  1047. // Execute and commit the block, update and save the state, and update the mempool.
  1048. // All calls to the proxyAppConn come here.
  1049. // NOTE: the block.AppHash wont reflect these txs until the next block
  1050. err := stateCopy.ApplyBlock(eventCache, cs.proxyAppConn, block, blockParts.Header(), cs.mempool)
  1051. if err != nil {
  1052. cs.Logger.Error("Error on ApplyBlock. Did the application crash? Please restart tendermint", "error", err)
  1053. return
  1054. }
  1055. fail.Fail() // XXX
  1056. // Fire event for new block.
  1057. // NOTE: If we fail before firing, these events will never fire
  1058. //
  1059. // TODO: Either
  1060. // * Fire before persisting state, in ApplyBlock
  1061. // * Fire on start up if we haven't written any new WAL msgs
  1062. // Both options mean we may fire more than once. Is that fine ?
  1063. types.FireEventNewBlock(cs.evsw, types.EventDataNewBlock{block})
  1064. types.FireEventNewBlockHeader(cs.evsw, types.EventDataNewBlockHeader{block.Header})
  1065. eventCache.Flush()
  1066. fail.Fail() // XXX
  1067. // NewHeightStep!
  1068. cs.updateToState(stateCopy)
  1069. fail.Fail() // XXX
  1070. // cs.StartTime is already set.
  1071. // Schedule Round0 to start soon.
  1072. cs.scheduleRound0(&cs.RoundState)
  1073. // By here,
  1074. // * cs.Height has been increment to height+1
  1075. // * cs.Step is now RoundStepNewHeight
  1076. // * cs.StartTime is set to when we will start round0.
  1077. return
  1078. }
  1079. //-----------------------------------------------------------------------------
  1080. func (cs *ConsensusState) defaultSetProposal(proposal *types.Proposal) error {
  1081. // Already have one
  1082. // TODO: possibly catch double proposals
  1083. if cs.Proposal != nil {
  1084. return nil
  1085. }
  1086. // Does not apply
  1087. if proposal.Height != cs.Height || proposal.Round != cs.Round {
  1088. return nil
  1089. }
  1090. // We don't care about the proposal if we're already in RoundStepCommit.
  1091. if RoundStepCommit <= cs.Step {
  1092. return nil
  1093. }
  1094. // Verify POLRound, which must be -1 or between 0 and proposal.Round exclusive.
  1095. if proposal.POLRound != -1 &&
  1096. (proposal.POLRound < 0 || proposal.Round <= proposal.POLRound) {
  1097. return ErrInvalidProposalPOLRound
  1098. }
  1099. // Verify signature
  1100. if !cs.Validators.GetProposer().PubKey.VerifyBytes(types.SignBytes(cs.state.ChainID, proposal), proposal.Signature) {
  1101. return ErrInvalidProposalSignature
  1102. }
  1103. cs.Proposal = proposal
  1104. cs.ProposalBlockParts = types.NewPartSetFromHeader(proposal.BlockPartsHeader)
  1105. return nil
  1106. }
  1107. // NOTE: block is not necessarily valid.
  1108. // Asynchronously triggers either enterPrevote (before we timeout of propose) or tryFinalizeCommit, once we have the full block.
  1109. func (cs *ConsensusState) addProposalBlockPart(height int, part *types.Part, verify bool) (added bool, err error) {
  1110. // Blocks might be reused, so round mismatch is OK
  1111. if cs.Height != height {
  1112. return false, nil
  1113. }
  1114. // We're not expecting a block part.
  1115. if cs.ProposalBlockParts == nil {
  1116. return false, nil // TODO: bad peer? Return error?
  1117. }
  1118. added, err = cs.ProposalBlockParts.AddPart(part, verify)
  1119. if err != nil {
  1120. return added, err
  1121. }
  1122. if added && cs.ProposalBlockParts.IsComplete() {
  1123. // Added and completed!
  1124. var n int
  1125. var err error
  1126. cs.ProposalBlock = wire.ReadBinary(&types.Block{}, cs.ProposalBlockParts.GetReader(), types.MaxBlockSize, &n, &err).(*types.Block)
  1127. // NOTE: it's possible to receive complete proposal blocks for future rounds without having the proposal
  1128. cs.Logger.Info("Received complete proposal block", "height", cs.ProposalBlock.Height, "hash", cs.ProposalBlock.Hash())
  1129. if cs.Step == RoundStepPropose && cs.isProposalComplete() {
  1130. // Move onto the next step
  1131. cs.enterPrevote(height, cs.Round)
  1132. } else if cs.Step == RoundStepCommit {
  1133. // If we're waiting on the proposal block...
  1134. cs.tryFinalizeCommit(height)
  1135. }
  1136. return true, err
  1137. }
  1138. return added, nil
  1139. }
  1140. // Attempt to add the vote. if its a duplicate signature, dupeout the validator
  1141. func (cs *ConsensusState) tryAddVote(vote *types.Vote, peerKey string) error {
  1142. _, err := cs.addVote(vote, peerKey)
  1143. if err != nil {
  1144. // If the vote height is off, we'll just ignore it,
  1145. // But if it's a conflicting sig, broadcast evidence tx for slashing.
  1146. // If it's otherwise invalid, punish peer.
  1147. if err == ErrVoteHeightMismatch {
  1148. return err
  1149. } else if _, ok := err.(*types.ErrVoteConflictingVotes); ok {
  1150. if peerKey == "" {
  1151. cs.Logger.Error("Found conflicting vote from ourselves. Did you unsafe_reset a validator?", "height", vote.Height, "round", vote.Round, "type", vote.Type)
  1152. return err
  1153. }
  1154. cs.Logger.Error("Found conflicting vote. Publish evidence (TODO)")
  1155. /* TODO
  1156. evidenceTx := &types.DupeoutTx{
  1157. Address: address,
  1158. VoteA: *errDupe.VoteA,
  1159. VoteB: *errDupe.VoteB,
  1160. }
  1161. cs.mempool.BroadcastTx(struct{???}{evidenceTx}) // shouldn't need to check returned err
  1162. */
  1163. return err
  1164. } else {
  1165. // Probably an invalid signature. Bad peer.
  1166. cs.Logger.Error("Error attempting to add vote", "error", err)
  1167. return ErrAddingVote
  1168. }
  1169. }
  1170. return nil
  1171. }
  1172. //-----------------------------------------------------------------------------
  1173. func (cs *ConsensusState) addVote(vote *types.Vote, peerKey string) (added bool, err error) {
  1174. cs.Logger.Debug("addVote", "voteHeight", vote.Height, "voteType", vote.Type, "csHeight", cs.Height)
  1175. // A precommit for the previous height?
  1176. // These come in while we wait timeoutCommit
  1177. if vote.Height+1 == cs.Height {
  1178. if !(cs.Step == RoundStepNewHeight && vote.Type == types.VoteTypePrecommit) {
  1179. // TODO: give the reason ..
  1180. // fmt.Errorf("tryAddVote: Wrong height, not a LastCommit straggler commit.")
  1181. return added, ErrVoteHeightMismatch
  1182. }
  1183. added, err = cs.LastCommit.AddVote(vote)
  1184. if added {
  1185. cs.Logger.Info(cmn.Fmt("Added to lastPrecommits: %v", cs.LastCommit.StringShort()))
  1186. types.FireEventVote(cs.evsw, types.EventDataVote{vote})
  1187. // if we can skip timeoutCommit and have all the votes now,
  1188. if cs.config.SkipTimeoutCommit && cs.LastCommit.HasAll() {
  1189. // go straight to new round (skip timeout commit)
  1190. // cs.scheduleTimeout(time.Duration(0), cs.Height, 0, RoundStepNewHeight)
  1191. cs.enterNewRound(cs.Height, 0)
  1192. }
  1193. }
  1194. return
  1195. }
  1196. // A prevote/precommit for this height?
  1197. if vote.Height == cs.Height {
  1198. height := cs.Height
  1199. added, err = cs.Votes.AddVote(vote, peerKey)
  1200. if added {
  1201. types.FireEventVote(cs.evsw, types.EventDataVote{vote})
  1202. switch vote.Type {
  1203. case types.VoteTypePrevote:
  1204. prevotes := cs.Votes.Prevotes(vote.Round)
  1205. cs.Logger.Info("Added to prevote", "vote", vote, "prevotes", prevotes.StringShort())
  1206. // First, unlock if prevotes is a valid POL.
  1207. // >> lockRound < POLRound <= unlockOrChangeLockRound (see spec)
  1208. // NOTE: If (lockRound < POLRound) but !(POLRound <= unlockOrChangeLockRound),
  1209. // we'll still enterNewRound(H,vote.R) and enterPrecommit(H,vote.R) to process it
  1210. // there.
  1211. if (cs.LockedBlock != nil) && (cs.LockedRound < vote.Round) && (vote.Round <= cs.Round) {
  1212. blockID, ok := prevotes.TwoThirdsMajority()
  1213. if ok && !cs.LockedBlock.HashesTo(blockID.Hash) {
  1214. cs.Logger.Info("Unlocking because of POL.", "lockedRound", cs.LockedRound, "POLRound", vote.Round)
  1215. cs.LockedRound = 0
  1216. cs.LockedBlock = nil
  1217. cs.LockedBlockParts = nil
  1218. types.FireEventUnlock(cs.evsw, cs.RoundStateEvent())
  1219. }
  1220. }
  1221. if cs.Round <= vote.Round && prevotes.HasTwoThirdsAny() {
  1222. // Round-skip over to PrevoteWait or goto Precommit.
  1223. cs.enterNewRound(height, vote.Round) // if the vote is ahead of us
  1224. if prevotes.HasTwoThirdsMajority() {
  1225. cs.enterPrecommit(height, vote.Round)
  1226. } else {
  1227. cs.enterPrevote(height, vote.Round) // if the vote is ahead of us
  1228. cs.enterPrevoteWait(height, vote.Round)
  1229. }
  1230. } else if cs.Proposal != nil && 0 <= cs.Proposal.POLRound && cs.Proposal.POLRound == vote.Round {
  1231. // If the proposal is now complete, enter prevote of cs.Round.
  1232. if cs.isProposalComplete() {
  1233. cs.enterPrevote(height, cs.Round)
  1234. }
  1235. }
  1236. case types.VoteTypePrecommit:
  1237. precommits := cs.Votes.Precommits(vote.Round)
  1238. cs.Logger.Info("Added to precommit", "vote", vote, "precommits", precommits.StringShort())
  1239. blockID, ok := precommits.TwoThirdsMajority()
  1240. if ok {
  1241. if len(blockID.Hash) == 0 {
  1242. cs.enterNewRound(height, vote.Round+1)
  1243. } else {
  1244. cs.enterNewRound(height, vote.Round)
  1245. cs.enterPrecommit(height, vote.Round)
  1246. cs.enterCommit(height, vote.Round)
  1247. if cs.config.SkipTimeoutCommit && precommits.HasAll() {
  1248. // if we have all the votes now,
  1249. // go straight to new round (skip timeout commit)
  1250. // cs.scheduleTimeout(time.Duration(0), cs.Height, 0, RoundStepNewHeight)
  1251. cs.enterNewRound(cs.Height, 0)
  1252. }
  1253. }
  1254. } else if cs.Round <= vote.Round && precommits.HasTwoThirdsAny() {
  1255. cs.enterNewRound(height, vote.Round)
  1256. cs.enterPrecommit(height, vote.Round)
  1257. cs.enterPrecommitWait(height, vote.Round)
  1258. }
  1259. default:
  1260. cmn.PanicSanity(cmn.Fmt("Unexpected vote type %X", vote.Type)) // Should not happen.
  1261. }
  1262. }
  1263. // Either duplicate, or error upon cs.Votes.AddByIndex()
  1264. return
  1265. } else {
  1266. err = ErrVoteHeightMismatch
  1267. }
  1268. // Height mismatch, bad peer?
  1269. cs.Logger.Info("Vote ignored and not added", "voteHeight", vote.Height, "csHeight", cs.Height, "err", err)
  1270. return
  1271. }
  1272. func (cs *ConsensusState) signVote(type_ byte, hash []byte, header types.PartSetHeader) (*types.Vote, error) {
  1273. addr := cs.privValidator.GetAddress()
  1274. valIndex, _ := cs.Validators.GetByAddress(addr)
  1275. vote := &types.Vote{
  1276. ValidatorAddress: addr,
  1277. ValidatorIndex: valIndex,
  1278. Height: cs.Height,
  1279. Round: cs.Round,
  1280. Type: type_,
  1281. BlockID: types.BlockID{hash, header},
  1282. }
  1283. err := cs.privValidator.SignVote(cs.state.ChainID, vote)
  1284. return vote, err
  1285. }
  1286. // sign the vote and publish on internalMsgQueue
  1287. func (cs *ConsensusState) signAddVote(type_ byte, hash []byte, header types.PartSetHeader) *types.Vote {
  1288. // if we don't have a key or we're not in the validator set, do nothing
  1289. if cs.privValidator == nil || !cs.Validators.HasAddress(cs.privValidator.GetAddress()) {
  1290. return nil
  1291. }
  1292. vote, err := cs.signVote(type_, hash, header)
  1293. if err == nil {
  1294. cs.sendInternalMessage(msgInfo{&VoteMessage{vote}, ""})
  1295. cs.Logger.Info("Signed and pushed vote", "height", cs.Height, "round", cs.Round, "vote", vote, "error", err)
  1296. return vote
  1297. } else {
  1298. //if !cs.replayMode {
  1299. cs.Logger.Error("Error signing vote", "height", cs.Height, "round", cs.Round, "vote", vote, "error", err)
  1300. //}
  1301. return nil
  1302. }
  1303. }
  1304. //---------------------------------------------------------
  1305. func CompareHRS(h1, r1 int, s1 RoundStepType, h2, r2 int, s2 RoundStepType) int {
  1306. if h1 < h2 {
  1307. return -1
  1308. } else if h1 > h2 {
  1309. return 1
  1310. }
  1311. if r1 < r2 {
  1312. return -1
  1313. } else if r1 > r2 {
  1314. return 1
  1315. }
  1316. if s1 < s2 {
  1317. return -1
  1318. } else if s1 > s2 {
  1319. return 1
  1320. }
  1321. return 0
  1322. }