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.

420 lines
14 KiB

8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
  1. package consensus
  2. import (
  3. "bytes"
  4. "errors"
  5. "fmt"
  6. "io"
  7. "reflect"
  8. "strconv"
  9. "strings"
  10. "time"
  11. abci "github.com/tendermint/abci/types"
  12. wire "github.com/tendermint/go-wire"
  13. auto "github.com/tendermint/tmlibs/autofile"
  14. cmn "github.com/tendermint/tmlibs/common"
  15. "github.com/tendermint/tmlibs/log"
  16. "github.com/tendermint/tendermint/proxy"
  17. sm "github.com/tendermint/tendermint/state"
  18. "github.com/tendermint/tendermint/types"
  19. )
  20. // Functionality to replay blocks and messages on recovery from a crash.
  21. // There are two general failure scenarios: failure during consensus, and failure while applying the block.
  22. // The former is handled by the WAL, the latter by the proxyApp Handshake on restart,
  23. // which ultimately hands off the work to the WAL.
  24. //-----------------------------------------
  25. // recover from failure during consensus
  26. // by replaying messages from the WAL
  27. // Unmarshal and apply a single message to the consensus state
  28. // as if it were received in receiveRoutine
  29. // Lines that start with "#" are ignored.
  30. // NOTE: receiveRoutine should not be running
  31. func (cs *ConsensusState) readReplayMessage(msgBytes []byte, newStepCh chan interface{}) error {
  32. // Skip over empty and meta lines
  33. if len(msgBytes) == 0 || msgBytes[0] == '#' {
  34. return nil
  35. }
  36. var err error
  37. var msg TimedWALMessage
  38. wire.ReadJSON(&msg, msgBytes, &err)
  39. if err != nil {
  40. fmt.Println("MsgBytes:", msgBytes, string(msgBytes))
  41. return fmt.Errorf("Error reading json data: %v", err)
  42. }
  43. // for logging
  44. switch m := msg.Msg.(type) {
  45. case types.EventDataRoundState:
  46. cs.Logger.Info("Replay: New Step", "height", m.Height, "round", m.Round, "step", m.Step)
  47. // these are playback checks
  48. ticker := time.After(time.Second * 2)
  49. if newStepCh != nil {
  50. select {
  51. case mi := <-newStepCh:
  52. m2 := mi.(types.EventDataRoundState)
  53. if m.Height != m2.Height || m.Round != m2.Round || m.Step != m2.Step {
  54. return fmt.Errorf("RoundState mismatch. Got %v; Expected %v", m2, m)
  55. }
  56. case <-ticker:
  57. return fmt.Errorf("Failed to read off newStepCh")
  58. }
  59. }
  60. case msgInfo:
  61. peerKey := m.PeerKey
  62. if peerKey == "" {
  63. peerKey = "local"
  64. }
  65. switch msg := m.Msg.(type) {
  66. case *ProposalMessage:
  67. p := msg.Proposal
  68. cs.Logger.Info("Replay: Proposal", "height", p.Height, "round", p.Round, "header",
  69. p.BlockPartsHeader, "pol", p.POLRound, "peer", peerKey)
  70. case *BlockPartMessage:
  71. cs.Logger.Info("Replay: BlockPart", "height", msg.Height, "round", msg.Round, "peer", peerKey)
  72. case *VoteMessage:
  73. v := msg.Vote
  74. cs.Logger.Info("Replay: Vote", "height", v.Height, "round", v.Round, "type", v.Type,
  75. "blockID", v.BlockID, "peer", peerKey)
  76. }
  77. cs.handleMsg(m)
  78. case timeoutInfo:
  79. cs.Logger.Info("Replay: Timeout", "height", m.Height, "round", m.Round, "step", m.Step, "dur", m.Duration)
  80. cs.handleTimeout(m, cs.RoundState)
  81. default:
  82. return fmt.Errorf("Replay: Unknown TimedWALMessage type: %v", reflect.TypeOf(msg.Msg))
  83. }
  84. return nil
  85. }
  86. // replay only those messages since the last block.
  87. // timeoutRoutine should run concurrently to read off tickChan
  88. func (cs *ConsensusState) catchupReplay(csHeight int) error {
  89. // set replayMode
  90. cs.replayMode = true
  91. defer func() { cs.replayMode = false }()
  92. // Ensure that ENDHEIGHT for this height doesn't exist
  93. // NOTE: This is just a sanity check. As far as we know things work fine without it,
  94. // and Handshake could reuse ConsensusState if it weren't for this check (since we can crash after writing ENDHEIGHT).
  95. gr, found, err := cs.wal.group.Search("#ENDHEIGHT: ", makeHeightSearchFunc(csHeight))
  96. if gr != nil {
  97. gr.Close()
  98. }
  99. if found {
  100. return errors.New(cmn.Fmt("WAL should not contain #ENDHEIGHT %d.", csHeight))
  101. }
  102. // Search for last height marker
  103. gr, found, err = cs.wal.group.Search("#ENDHEIGHT: ", makeHeightSearchFunc(csHeight-1))
  104. if err == io.EOF {
  105. cs.Logger.Error("Replay: wal.group.Search returned EOF", "#ENDHEIGHT", csHeight-1)
  106. // if we upgraded from 0.9 to 0.9.1, we may have #HEIGHT instead
  107. // TODO (0.10.0): remove this
  108. gr, found, err = cs.wal.group.Search("#HEIGHT: ", makeHeightSearchFunc(csHeight))
  109. if err == io.EOF {
  110. cs.Logger.Error("Replay: wal.group.Search returned EOF", "#HEIGHT", csHeight)
  111. return nil
  112. } else if err != nil {
  113. return err
  114. }
  115. } else if err != nil {
  116. return err
  117. } else {
  118. defer gr.Close()
  119. }
  120. if !found {
  121. // if we upgraded from 0.9 to 0.9.1, we may have #HEIGHT instead
  122. // TODO (0.10.0): remove this
  123. gr, _, err = cs.wal.group.Search("#HEIGHT: ", makeHeightSearchFunc(csHeight))
  124. if err == io.EOF {
  125. cs.Logger.Error("Replay: wal.group.Search returned EOF", "#HEIGHT", csHeight)
  126. return nil
  127. } else if err != nil {
  128. return err
  129. } else {
  130. defer gr.Close()
  131. }
  132. // TODO (0.10.0): uncomment
  133. // return errors.New(cmn.Fmt("Cannot replay height %d. WAL does not contain #ENDHEIGHT for %d.", csHeight, csHeight-1))
  134. }
  135. cs.Logger.Info("Catchup by replaying consensus messages", "height", csHeight)
  136. for {
  137. line, err := gr.ReadLine()
  138. if err != nil {
  139. if err == io.EOF {
  140. break
  141. } else {
  142. return err
  143. }
  144. }
  145. // NOTE: since the priv key is set when the msgs are received
  146. // it will attempt to eg double sign but we can just ignore it
  147. // since the votes will be replayed and we'll get to the next step
  148. if err := cs.readReplayMessage([]byte(line), nil); err != nil {
  149. return err
  150. }
  151. }
  152. cs.Logger.Info("Replay: Done")
  153. return nil
  154. }
  155. //--------------------------------------------------------------------------------
  156. // Parses marker lines of the form:
  157. // #ENDHEIGHT: 12345
  158. func makeHeightSearchFunc(height int) auto.SearchFunc {
  159. return func(line string) (int, error) {
  160. line = strings.TrimRight(line, "\n")
  161. parts := strings.Split(line, " ")
  162. if len(parts) != 2 {
  163. return -1, errors.New("Line did not have 2 parts")
  164. }
  165. i, err := strconv.Atoi(parts[1])
  166. if err != nil {
  167. return -1, errors.New("Failed to parse INFO: " + err.Error())
  168. }
  169. if height < i {
  170. return 1, nil
  171. } else if height == i {
  172. return 0, nil
  173. } else {
  174. return -1, nil
  175. }
  176. }
  177. }
  178. //----------------------------------------------
  179. // Recover from failure during block processing
  180. // by handshaking with the app to figure out where
  181. // we were last and using the WAL to recover there
  182. type Handshaker struct {
  183. state *sm.State
  184. store types.BlockStore
  185. logger log.Logger
  186. nBlocks int // number of blocks applied to the state
  187. }
  188. func NewHandshaker(state *sm.State, store types.BlockStore) *Handshaker {
  189. return &Handshaker{state, store, log.NewNopLogger(), 0}
  190. }
  191. func (h *Handshaker) SetLogger(l log.Logger) {
  192. h.logger = l
  193. }
  194. func (h *Handshaker) NBlocks() int {
  195. return h.nBlocks
  196. }
  197. // TODO: retry the handshake/replay if it fails ?
  198. func (h *Handshaker) Handshake(proxyApp proxy.AppConns) error {
  199. // handshake is done via info request on the query conn
  200. res, err := proxyApp.Query().InfoSync()
  201. if err != nil {
  202. return errors.New(cmn.Fmt("Error calling Info: %v", err))
  203. }
  204. blockHeight := int(res.LastBlockHeight) // XXX: beware overflow
  205. appHash := res.LastBlockAppHash
  206. h.logger.Info("ABCI Handshake", "appHeight", blockHeight, "appHash", fmt.Sprintf("%X", appHash))
  207. // TODO: check version
  208. // replay blocks up to the latest in the blockstore
  209. _, err = h.ReplayBlocks(appHash, blockHeight, proxyApp)
  210. if err != nil {
  211. return errors.New(cmn.Fmt("Error on replay: %v", err))
  212. }
  213. h.logger.Info("Completed ABCI Handshake - Tendermint and App are synced", "appHeight", blockHeight, "appHash", fmt.Sprintf("%X", appHash))
  214. // TODO: (on restart) replay mempool
  215. return nil
  216. }
  217. // Replay all blocks since appBlockHeight and ensure the result matches the current state.
  218. // Returns the final AppHash or an error
  219. func (h *Handshaker) ReplayBlocks(appHash []byte, appBlockHeight int, proxyApp proxy.AppConns) ([]byte, error) {
  220. storeBlockHeight := h.store.Height()
  221. stateBlockHeight := h.state.LastBlockHeight
  222. h.logger.Info("ABCI Replay Blocks", "appHeight", appBlockHeight, "storeHeight", storeBlockHeight, "stateHeight", stateBlockHeight)
  223. // If appBlockHeight == 0 it means that we are at genesis and hence should send InitChain
  224. if appBlockHeight == 0 {
  225. validators := types.TM2PB.Validators(h.state.Validators)
  226. proxyApp.Consensus().InitChainSync(validators)
  227. }
  228. // First handle edge cases and constraints on the storeBlockHeight
  229. if storeBlockHeight == 0 {
  230. return appHash, h.checkAppHash(appHash)
  231. } else if storeBlockHeight < appBlockHeight {
  232. // the app should never be ahead of the store (but this is under app's control)
  233. return appHash, sm.ErrAppBlockHeightTooHigh{storeBlockHeight, appBlockHeight}
  234. } else if storeBlockHeight < stateBlockHeight {
  235. // the state should never be ahead of the store (this is under tendermint's control)
  236. cmn.PanicSanity(cmn.Fmt("StateBlockHeight (%d) > StoreBlockHeight (%d)", stateBlockHeight, storeBlockHeight))
  237. } else if storeBlockHeight > stateBlockHeight+1 {
  238. // store should be at most one ahead of the state (this is under tendermint's control)
  239. cmn.PanicSanity(cmn.Fmt("StoreBlockHeight (%d) > StateBlockHeight + 1 (%d)", storeBlockHeight, stateBlockHeight+1))
  240. }
  241. // Now either store is equal to state, or one ahead.
  242. // For each, consider all cases of where the app could be, given app <= store
  243. if storeBlockHeight == stateBlockHeight {
  244. // Tendermint ran Commit and saved the state.
  245. // Either the app is asking for replay, or we're all synced up.
  246. if appBlockHeight < storeBlockHeight {
  247. // the app is behind, so replay blocks, but no need to go through WAL (state is already synced to store)
  248. return h.replayBlocks(proxyApp, appBlockHeight, storeBlockHeight, false)
  249. } else if appBlockHeight == storeBlockHeight {
  250. // We're good!
  251. return appHash, h.checkAppHash(appHash)
  252. }
  253. } else if storeBlockHeight == stateBlockHeight+1 {
  254. // We saved the block in the store but haven't updated the state,
  255. // so we'll need to replay a block using the WAL.
  256. if appBlockHeight < stateBlockHeight {
  257. // the app is further behind than it should be, so replay blocks
  258. // but leave the last block to go through the WAL
  259. return h.replayBlocks(proxyApp, appBlockHeight, storeBlockHeight, true)
  260. } else if appBlockHeight == stateBlockHeight {
  261. // We haven't run Commit (both the state and app are one block behind),
  262. // so replayBlock with the real app.
  263. // NOTE: We could instead use the cs.WAL on cs.Start,
  264. // but we'd have to allow the WAL to replay a block that wrote it's ENDHEIGHT
  265. h.logger.Info("Replay last block using real app")
  266. return h.replayBlock(storeBlockHeight, proxyApp.Consensus())
  267. } else if appBlockHeight == storeBlockHeight {
  268. // We ran Commit, but didn't save the state, so replayBlock with mock app
  269. abciResponses := h.state.LoadABCIResponses()
  270. mockApp := newMockProxyApp(appHash, abciResponses)
  271. h.logger.Info("Replay last block using mock app")
  272. return h.replayBlock(storeBlockHeight, mockApp)
  273. }
  274. }
  275. cmn.PanicSanity("Should never happen")
  276. return nil, nil
  277. }
  278. func (h *Handshaker) replayBlocks(proxyApp proxy.AppConns, appBlockHeight, storeBlockHeight int, mutateState bool) ([]byte, error) {
  279. // App is further behind than it should be, so we need to replay blocks.
  280. // We replay all blocks from appBlockHeight+1.
  281. // Note that we don't have an old version of the state,
  282. // so we by-pass state validation/mutation using sm.ExecCommitBlock.
  283. // If mutateState == true, the final block is replayed with h.replayBlock()
  284. var appHash []byte
  285. var err error
  286. finalBlock := storeBlockHeight
  287. if mutateState {
  288. finalBlock -= 1
  289. }
  290. for i := appBlockHeight + 1; i <= finalBlock; i++ {
  291. h.logger.Info("Applying block", "height", i)
  292. block := h.store.LoadBlock(i)
  293. appHash, err = sm.ExecCommitBlock(proxyApp.Consensus(), block, h.logger)
  294. if err != nil {
  295. return nil, err
  296. }
  297. h.nBlocks += 1
  298. }
  299. if mutateState {
  300. // sync the final block
  301. return h.replayBlock(storeBlockHeight, proxyApp.Consensus())
  302. }
  303. return appHash, h.checkAppHash(appHash)
  304. }
  305. // ApplyBlock on the proxyApp with the last block.
  306. func (h *Handshaker) replayBlock(height int, proxyApp proxy.AppConnConsensus) ([]byte, error) {
  307. mempool := types.MockMempool{}
  308. var eventCache types.Fireable // nil
  309. block := h.store.LoadBlock(height)
  310. meta := h.store.LoadBlockMeta(height)
  311. if err := h.state.ApplyBlock(eventCache, proxyApp, block, meta.BlockID.PartsHeader, mempool); err != nil {
  312. return nil, err
  313. }
  314. h.nBlocks += 1
  315. return h.state.AppHash, nil
  316. }
  317. func (h *Handshaker) checkAppHash(appHash []byte) error {
  318. if !bytes.Equal(h.state.AppHash, appHash) {
  319. panic(errors.New(cmn.Fmt("Tendermint state.AppHash does not match AppHash after replay. Got %X, expected %X", appHash, h.state.AppHash)).Error())
  320. return nil
  321. }
  322. return nil
  323. }
  324. //--------------------------------------------------------------------------------
  325. // mockProxyApp uses ABCIResponses to give the right results
  326. // Useful because we don't want to call Commit() twice for the same block on the real app.
  327. func newMockProxyApp(appHash []byte, abciResponses *sm.ABCIResponses) proxy.AppConnConsensus {
  328. clientCreator := proxy.NewLocalClientCreator(&mockProxyApp{
  329. appHash: appHash,
  330. abciResponses: abciResponses,
  331. })
  332. cli, _ := clientCreator.NewABCIClient()
  333. cli.Start()
  334. return proxy.NewAppConnConsensus(cli)
  335. }
  336. type mockProxyApp struct {
  337. abci.BaseApplication
  338. appHash []byte
  339. txCount int
  340. abciResponses *sm.ABCIResponses
  341. }
  342. func (mock *mockProxyApp) DeliverTx(tx []byte) abci.Result {
  343. r := mock.abciResponses.DeliverTx[mock.txCount]
  344. mock.txCount += 1
  345. return abci.Result{
  346. r.Code,
  347. r.Data,
  348. r.Log,
  349. }
  350. }
  351. func (mock *mockProxyApp) EndBlock(height uint64) abci.ResponseEndBlock {
  352. mock.txCount = 0
  353. return mock.abciResponses.EndBlock
  354. }
  355. func (mock *mockProxyApp) Commit() abci.Result {
  356. return abci.NewResultOK(mock.appHash, "")
  357. }