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.

1447 lines
48 KiB

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