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.

415 lines
13 KiB

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. auto "github.com/tendermint/tmlibs/autofile"
  13. . "github.com/tendermint/tmlibs/common"
  14. cfg "github.com/tendermint/go-config"
  15. "github.com/tendermint/go-wire"
  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. log.Notice("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. log.Notice("Replay: Proposal", "height", p.Height, "round", p.Round, "header",
  69. p.BlockPartsHeader, "pol", p.POLRound, "peer", peerKey)
  70. case *BlockPartMessage:
  71. log.Notice("Replay: BlockPart", "height", msg.Height, "round", msg.Round, "peer", peerKey)
  72. case *VoteMessage:
  73. v := msg.Vote
  74. log.Notice("Replay: Vote", "height", v.Height, "round", v.Round, "type", v.Type,
  75. "blockID", v.BlockID, "peer", peerKey)
  76. }
  77. cs.handleMsg(m, cs.RoundState)
  78. case timeoutInfo:
  79. log.Notice("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(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. log.Warn("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. log.Warn("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, found, err = cs.wal.group.Search("#HEIGHT: ", makeHeightSearchFunc(csHeight))
  124. if err == io.EOF {
  125. log.Warn("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(Fmt("Cannot replay height %d. WAL does not contain #ENDHEIGHT for %d.", csHeight, csHeight-1))
  134. }
  135. log.Notice("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. log.Notice("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. config cfg.Config
  184. state *sm.State
  185. store types.BlockStore
  186. nBlocks int // number of blocks applied to the state
  187. }
  188. func NewHandshaker(config cfg.Config, state *sm.State, store types.BlockStore) *Handshaker {
  189. return &Handshaker{config, state, store, 0}
  190. }
  191. func (h *Handshaker) NBlocks() int {
  192. return h.nBlocks
  193. }
  194. var ErrReplayLastBlockTimeout = errors.New("Timed out waiting for last block to be replayed")
  195. // TODO: retry the handshake/replay if it fails ?
  196. func (h *Handshaker) Handshake(proxyApp proxy.AppConns) error {
  197. // handshake is done via info request on the query conn
  198. res, err := proxyApp.Query().InfoSync()
  199. if err != nil {
  200. return errors.New(Fmt("Error calling Info: %v", err))
  201. }
  202. blockHeight := int(res.LastBlockHeight) // XXX: beware overflow
  203. appHash := res.LastBlockAppHash
  204. log.Notice("ABCI Handshake", "appHeight", blockHeight, "appHash", appHash)
  205. // TODO: check version
  206. // replay blocks up to the latest in the blockstore
  207. _, err = h.ReplayBlocks(appHash, blockHeight, proxyApp)
  208. if err == ErrReplayLastBlockTimeout {
  209. log.Warn("Failed to sync via handshake. Trying other means. If they fail, please increase the timeout_handshake parameter")
  210. return nil
  211. } else if err != nil {
  212. return errors.New(Fmt("Error on replay: %v", err))
  213. }
  214. log.Notice("Completed ABCI Handshake - Tendermint and App are synced", "appHeight", blockHeight, "appHash", appHash)
  215. // TODO: (on restart) replay mempool
  216. return nil
  217. }
  218. // Replay all blocks since appBlockHeight and ensure the result matches the current state.
  219. // Returns the final AppHash or an error
  220. func (h *Handshaker) ReplayBlocks(appHash []byte, appBlockHeight int, proxyApp proxy.AppConns) ([]byte, error) {
  221. storeBlockHeight := h.store.Height()
  222. stateBlockHeight := h.state.LastBlockHeight
  223. log.Notice("ABCI Replay Blocks", "appHeight", appBlockHeight, "storeHeight", storeBlockHeight, "stateHeight", stateBlockHeight)
  224. // First handle edge cases and constraints on the storeBlockHeight
  225. if storeBlockHeight == 0 {
  226. return appHash, h.checkAppHash(appHash)
  227. } else if storeBlockHeight < appBlockHeight {
  228. // the app should never be ahead of the store (but this is under app's control)
  229. return appHash, sm.ErrAppBlockHeightTooHigh{storeBlockHeight, appBlockHeight}
  230. } else if storeBlockHeight < stateBlockHeight {
  231. // the state should never be ahead of the store (this is under tendermint's control)
  232. PanicSanity(Fmt("StateBlockHeight (%d) > StoreBlockHeight (%d)", stateBlockHeight, storeBlockHeight))
  233. } else if storeBlockHeight > stateBlockHeight+1 {
  234. // store should be at most one ahead of the state (this is under tendermint's control)
  235. PanicSanity(Fmt("StoreBlockHeight (%d) > StateBlockHeight + 1 (%d)", storeBlockHeight, stateBlockHeight+1))
  236. }
  237. // Now either store is equal to state, or one ahead.
  238. // For each, consider all cases of where the app could be, given app <= store
  239. if storeBlockHeight == stateBlockHeight {
  240. // Tendermint ran Commit and saved the state.
  241. // Either the app is asking for replay, or we're all synced up.
  242. if appBlockHeight < storeBlockHeight {
  243. // the app is behind, so replay blocks, but no need to go through WAL (state is already synced to store)
  244. return h.replayBlocks(proxyApp, appBlockHeight, storeBlockHeight, false)
  245. } else if appBlockHeight == storeBlockHeight {
  246. // We're good!
  247. return appHash, h.checkAppHash(appHash)
  248. }
  249. } else if storeBlockHeight == stateBlockHeight+1 {
  250. // We saved the block in the store but haven't updated the state,
  251. // so we'll need to replay a block using the WAL.
  252. if appBlockHeight < stateBlockHeight {
  253. // the app is further behind than it should be, so replay blocks
  254. // but leave the last block to go through the WAL
  255. return h.replayBlocks(proxyApp, appBlockHeight, storeBlockHeight, true)
  256. } else if appBlockHeight == stateBlockHeight {
  257. // We haven't run Commit (both the state and app are one block behind),
  258. // so replayBlock with the real app.
  259. // NOTE: We could instead use the cs.WAL on cs.Start,
  260. // but we'd have to allow the WAL to replay a block that wrote it's ENDHEIGHT
  261. log.Info("Replay last block using real app")
  262. return h.replayBlock(storeBlockHeight, proxyApp.Consensus())
  263. } else if appBlockHeight == storeBlockHeight {
  264. // We ran Commit, but didn't save the state, so replayBlock with mock app
  265. abciResponses := h.state.LoadABCIResponses()
  266. mockApp := newMockProxyApp(appHash, abciResponses)
  267. log.Info("Replay last block using mock app")
  268. return h.replayBlock(storeBlockHeight, mockApp)
  269. }
  270. }
  271. PanicSanity("Should never happen")
  272. return nil, nil
  273. }
  274. func (h *Handshaker) replayBlocks(proxyApp proxy.AppConns, appBlockHeight, storeBlockHeight int, mutateState bool) ([]byte, error) {
  275. // App is further behind than it should be, so we need to replay blocks.
  276. // We replay all blocks from appBlockHeight+1.
  277. // Note that we don't have an old version of the state,
  278. // so we by-pass state validation/mutation using sm.ExecCommitBlock.
  279. // If mutateState == true, the final block is replayed with h.replayBlock()
  280. var appHash []byte
  281. var err error
  282. finalBlock := storeBlockHeight
  283. if mutateState {
  284. finalBlock -= 1
  285. }
  286. for i := appBlockHeight + 1; i <= finalBlock; i++ {
  287. log.Info("Applying block", "height", i)
  288. block := h.store.LoadBlock(i)
  289. appHash, err = sm.ExecCommitBlock(proxyApp.Consensus(), block)
  290. if err != nil {
  291. return nil, err
  292. }
  293. h.nBlocks += 1
  294. }
  295. if mutateState {
  296. // sync the final block
  297. return h.replayBlock(storeBlockHeight, proxyApp.Consensus())
  298. }
  299. return appHash, h.checkAppHash(appHash)
  300. }
  301. // ApplyBlock on the proxyApp with the last block.
  302. func (h *Handshaker) replayBlock(height int, proxyApp proxy.AppConnConsensus) ([]byte, error) {
  303. mempool := types.MockMempool{}
  304. var eventCache types.Fireable // nil
  305. block := h.store.LoadBlock(height)
  306. meta := h.store.LoadBlockMeta(height)
  307. if err := h.state.ApplyBlock(eventCache, proxyApp, block, meta.BlockID.PartsHeader, mempool); err != nil {
  308. return nil, err
  309. }
  310. h.nBlocks += 1
  311. return h.state.AppHash, nil
  312. }
  313. func (h *Handshaker) checkAppHash(appHash []byte) error {
  314. if !bytes.Equal(h.state.AppHash, appHash) {
  315. panic(errors.New(Fmt("Tendermint state.AppHash does not match AppHash after replay. Got %X, expected %X", appHash, h.state.AppHash)).Error())
  316. return nil
  317. }
  318. return nil
  319. }
  320. //--------------------------------------------------------------------------------
  321. // mockProxyApp uses ABCIResponses to give the right results
  322. // Useful because we don't want to call Commit() twice for the same block on the real app.
  323. func newMockProxyApp(appHash []byte, abciResponses *sm.ABCIResponses) proxy.AppConnConsensus {
  324. clientCreator := proxy.NewLocalClientCreator(&mockProxyApp{
  325. appHash: appHash,
  326. abciResponses: abciResponses,
  327. })
  328. cli, _ := clientCreator.NewABCIClient()
  329. return proxy.NewAppConnConsensus(cli)
  330. }
  331. type mockProxyApp struct {
  332. abci.BaseApplication
  333. appHash []byte
  334. txCount int
  335. abciResponses *sm.ABCIResponses
  336. }
  337. func (mock *mockProxyApp) DeliverTx(tx []byte) abci.Result {
  338. r := mock.abciResponses.DeliverTx[mock.txCount]
  339. mock.txCount += 1
  340. return abci.Result{
  341. r.Code,
  342. r.Data,
  343. r.Log,
  344. }
  345. }
  346. func (mock *mockProxyApp) EndBlock(height uint64) abci.ResponseEndBlock {
  347. mock.txCount = 0
  348. return mock.abciResponses.EndBlock
  349. }
  350. func (mock *mockProxyApp) Commit() abci.Result {
  351. return abci.NewResultOK(mock.appHash, "")
  352. }