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.

1544 lines
52 KiB

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