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.

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