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.

424 lines
14 KiB

8 years ago
7 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 state
  2. import (
  3. "errors"
  4. "fmt"
  5. fail "github.com/ebuchman/fail-test"
  6. abci "github.com/tendermint/abci/types"
  7. crypto "github.com/tendermint/go-crypto"
  8. "github.com/tendermint/tendermint/proxy"
  9. "github.com/tendermint/tendermint/types"
  10. dbm "github.com/tendermint/tmlibs/db"
  11. "github.com/tendermint/tmlibs/log"
  12. )
  13. //-----------------------------------------------------------------------------
  14. // BlockExecutor handles block execution and state updates.
  15. // It exposes ApplyBlock(), which validates & executes the block, updates state w/ ABCI responses,
  16. // then commits and updates the mempool atomically, then saves state.
  17. // BlockExecutor provides the context and accessories for properly executing a block.
  18. type BlockExecutor struct {
  19. // save state, validators, consensus params, abci responses here
  20. db dbm.DB
  21. // execute the app against this
  22. proxyApp proxy.AppConnConsensus
  23. // events
  24. eventBus types.BlockEventPublisher
  25. // update these with block results after commit
  26. mempool types.Mempool
  27. evpool types.EvidencePool
  28. logger log.Logger
  29. }
  30. // NewBlockExecutor returns a new BlockExecutor with a NopEventBus.
  31. // Call SetEventBus to provide one.
  32. func NewBlockExecutor(db dbm.DB, logger log.Logger, proxyApp proxy.AppConnConsensus,
  33. mempool types.Mempool, evpool types.EvidencePool) *BlockExecutor {
  34. return &BlockExecutor{
  35. db: db,
  36. proxyApp: proxyApp,
  37. eventBus: types.NopEventBus{},
  38. mempool: mempool,
  39. evpool: evpool,
  40. logger: logger,
  41. }
  42. }
  43. // SetEventBus - sets the event bus for publishing block related events.
  44. // If not called, it defaults to types.NopEventBus.
  45. func (blockExec *BlockExecutor) SetEventBus(eventBus types.BlockEventPublisher) {
  46. blockExec.eventBus = eventBus
  47. }
  48. // ValidateBlock validates the given block against the given state.
  49. // If the block is invalid, it returns an error.
  50. // Validation does not mutate state, but does require historical information from the stateDB,
  51. // ie. to verify evidence from a validator at an old height.
  52. func (blockExec *BlockExecutor) ValidateBlock(s State, block *types.Block) error {
  53. return validateBlock(blockExec.db, s, block)
  54. }
  55. // ApplyBlock validates the block against the state, executes it against the app,
  56. // fires the relevant events, commits the app, and saves the new state and responses.
  57. // It's the only function that needs to be called
  58. // from outside this package to process and commit an entire block.
  59. // It takes a blockID to avoid recomputing the parts hash.
  60. func (blockExec *BlockExecutor) ApplyBlock(s State, blockID types.BlockID, block *types.Block) (State, error) {
  61. if err := blockExec.ValidateBlock(s, block); err != nil {
  62. return s, ErrInvalidBlock(err)
  63. }
  64. abciResponses, err := execBlockOnProxyApp(blockExec.logger, blockExec.proxyApp, block)
  65. if err != nil {
  66. return s, ErrProxyAppConn(err)
  67. }
  68. fail.Fail() // XXX
  69. // save the results before we commit
  70. saveABCIResponses(blockExec.db, block.Height, abciResponses)
  71. fail.Fail() // XXX
  72. // update the state with the block and responses
  73. s, err = updateState(s, blockID, block.Header, abciResponses)
  74. if err != nil {
  75. return s, fmt.Errorf("Commit failed for application: %v", err)
  76. }
  77. // lock mempool, commit state, update mempoool
  78. appHash, err := blockExec.Commit(block)
  79. if err != nil {
  80. return s, fmt.Errorf("Commit failed for application: %v", err)
  81. }
  82. fail.Fail() // XXX
  83. // update the app hash and save the state
  84. s.AppHash = appHash
  85. SaveState(blockExec.db, s)
  86. fail.Fail() // XXX
  87. // Update evpool now that state is saved
  88. // TODO: handle the crash/recover scenario
  89. // ie. (may need to call Update for last block)
  90. blockExec.evpool.Update(block)
  91. // events are fired after everything else
  92. // NOTE: if we crash between Commit and Save, events wont be fired during replay
  93. fireEvents(blockExec.logger, blockExec.eventBus, block, abciResponses)
  94. return s, nil
  95. }
  96. // Commit locks the mempool, runs the ABCI Commit message, and updates the mempool.
  97. // It returns the result of calling abci.Commit (the AppHash), and an error.
  98. // The Mempool must be locked during commit and update because state is typically reset on Commit and old txs must be replayed
  99. // against committed state before new txs are run in the mempool, lest they be invalid.
  100. func (blockExec *BlockExecutor) Commit(block *types.Block) ([]byte, error) {
  101. blockExec.mempool.Lock()
  102. defer blockExec.mempool.Unlock()
  103. // while mempool is Locked, flush to ensure all async requests have completed
  104. // in the ABCI app before Commit.
  105. err := blockExec.mempool.FlushAppConn()
  106. if err != nil {
  107. blockExec.logger.Error("Client error during mempool.FlushAppConn", "err", err)
  108. return nil, err
  109. }
  110. // Commit block, get hash back
  111. res, err := blockExec.proxyApp.CommitSync()
  112. if err != nil {
  113. blockExec.logger.Error("Client error during proxyAppConn.CommitSync", "err", err)
  114. return nil, err
  115. }
  116. // ResponseCommit has no error code - just data
  117. blockExec.logger.Info("Committed state", "height", block.Height, "txs", block.NumTxs, "appHash", res.Data)
  118. // Update mempool.
  119. if err := blockExec.mempool.Update(block.Height, block.Txs); err != nil {
  120. return nil, err
  121. }
  122. return res.Data, nil
  123. }
  124. //---------------------------------------------------------
  125. // Helper functions for executing blocks and updating state
  126. // Executes block's transactions on proxyAppConn.
  127. // Returns a list of transaction results and updates to the validator set
  128. func execBlockOnProxyApp(logger log.Logger, proxyAppConn proxy.AppConnConsensus, block *types.Block) (*ABCIResponses, error) {
  129. var validTxs, invalidTxs = 0, 0
  130. txIndex := 0
  131. abciResponses := NewABCIResponses(block)
  132. // Execute transactions and get hash
  133. proxyCb := func(req *abci.Request, res *abci.Response) {
  134. switch r := res.Value.(type) {
  135. case *abci.Response_DeliverTx:
  136. // TODO: make use of res.Log
  137. // TODO: make use of this info
  138. // Blocks may include invalid txs.
  139. txRes := r.DeliverTx
  140. if txRes.Code == abci.CodeTypeOK {
  141. validTxs++
  142. } else {
  143. logger.Debug("Invalid tx", "code", txRes.Code, "log", txRes.Log)
  144. invalidTxs++
  145. }
  146. abciResponses.DeliverTx[txIndex] = txRes
  147. txIndex++
  148. }
  149. }
  150. proxyAppConn.SetResponseCallback(proxyCb)
  151. // determine which validators did not sign last block
  152. absentVals := make([]int32, 0)
  153. for valI, vote := range block.LastCommit.Precommits {
  154. if vote == nil {
  155. absentVals = append(absentVals, int32(valI))
  156. }
  157. }
  158. // TODO: determine which validators were byzantine
  159. byzantineVals := make([]abci.Evidence, len(block.Evidence.Evidence))
  160. for i, ev := range block.Evidence.Evidence {
  161. byzantineVals[i] = abci.Evidence{
  162. PubKey: ev.Address(), // XXX
  163. Height: ev.Height(),
  164. }
  165. }
  166. // Begin block
  167. _, err := proxyAppConn.BeginBlockSync(abci.RequestBeginBlock{
  168. Hash: block.Hash(),
  169. Header: types.TM2PB.Header(block.Header),
  170. AbsentValidators: absentVals,
  171. ByzantineValidators: byzantineVals,
  172. })
  173. if err != nil {
  174. logger.Error("Error in proxyAppConn.BeginBlock", "err", err)
  175. return nil, err
  176. }
  177. // Run txs of block
  178. for _, tx := range block.Txs {
  179. proxyAppConn.DeliverTxAsync(tx)
  180. if err := proxyAppConn.Error(); err != nil {
  181. return nil, err
  182. }
  183. }
  184. // End block
  185. abciResponses.EndBlock, err = proxyAppConn.EndBlockSync(abci.RequestEndBlock{block.Height})
  186. if err != nil {
  187. logger.Error("Error in proxyAppConn.EndBlock", "err", err)
  188. return nil, err
  189. }
  190. logger.Info("Executed block", "height", block.Height, "validTxs", validTxs, "invalidTxs", invalidTxs)
  191. valUpdates := abciResponses.EndBlock.ValidatorUpdates
  192. if len(valUpdates) > 0 {
  193. logger.Info("Updates to validators", "updates", abci.ValidatorsString(valUpdates))
  194. }
  195. return abciResponses, nil
  196. }
  197. func updateValidators(currentSet *types.ValidatorSet, updates []abci.Validator) error {
  198. // If more or equal than 1/3 of total voting power changed in one block, then
  199. // a light client could never prove the transition externally. See
  200. // ./lite/doc.go for details on how a light client tracks validators.
  201. vp23, err := changeInVotingPowerMoreOrEqualToOneThird(currentSet, updates)
  202. if err != nil {
  203. return err
  204. }
  205. if vp23 {
  206. return errors.New("the change in voting power must be strictly less than 1/3")
  207. }
  208. for _, v := range updates {
  209. pubkey, err := crypto.PubKeyFromBytes(v.PubKey) // NOTE: expects go-wire encoded pubkey
  210. if err != nil {
  211. return err
  212. }
  213. address := pubkey.Address()
  214. power := int64(v.Power)
  215. // mind the overflow from int64
  216. if power < 0 {
  217. return fmt.Errorf("Power (%d) overflows int64", v.Power)
  218. }
  219. _, val := currentSet.GetByAddress(address)
  220. if val == nil {
  221. // add val
  222. added := currentSet.Add(types.NewValidator(pubkey, power))
  223. if !added {
  224. return fmt.Errorf("Failed to add new validator %X with voting power %d", address, power)
  225. }
  226. } else if v.Power == 0 {
  227. // remove val
  228. _, removed := currentSet.Remove(address)
  229. if !removed {
  230. return fmt.Errorf("Failed to remove validator %X", address)
  231. }
  232. } else {
  233. // update val
  234. val.VotingPower = power
  235. updated := currentSet.Update(val)
  236. if !updated {
  237. return fmt.Errorf("Failed to update validator %X with voting power %d", address, power)
  238. }
  239. }
  240. }
  241. return nil
  242. }
  243. func changeInVotingPowerMoreOrEqualToOneThird(currentSet *types.ValidatorSet, updates []abci.Validator) (bool, error) {
  244. threshold := currentSet.TotalVotingPower() * 1 / 3
  245. acc := int64(0)
  246. for _, v := range updates {
  247. pubkey, err := crypto.PubKeyFromBytes(v.PubKey) // NOTE: expects go-wire encoded pubkey
  248. if err != nil {
  249. return false, err
  250. }
  251. address := pubkey.Address()
  252. power := int64(v.Power)
  253. // mind the overflow from int64
  254. if power < 0 {
  255. return false, fmt.Errorf("Power (%d) overflows int64", v.Power)
  256. }
  257. _, val := currentSet.GetByAddress(address)
  258. if val == nil {
  259. acc += power
  260. } else {
  261. np := val.VotingPower - power
  262. if np < 0 {
  263. np = -np
  264. }
  265. acc += np
  266. }
  267. if acc >= threshold {
  268. return true, nil
  269. }
  270. }
  271. return false, nil
  272. }
  273. // updateState returns a new State updated according to the header and responses.
  274. func updateState(s State, blockID types.BlockID, header *types.Header,
  275. abciResponses *ABCIResponses) (State, error) {
  276. // copy the valset so we can apply changes from EndBlock
  277. // and update s.LastValidators and s.Validators
  278. prevValSet := s.Validators.Copy()
  279. nextValSet := prevValSet.Copy()
  280. // update the validator set with the latest abciResponses
  281. lastHeightValsChanged := s.LastHeightValidatorsChanged
  282. if len(abciResponses.EndBlock.ValidatorUpdates) > 0 {
  283. err := updateValidators(nextValSet, abciResponses.EndBlock.ValidatorUpdates)
  284. if err != nil {
  285. return s, fmt.Errorf("Error changing validator set: %v", err)
  286. }
  287. // change results from this height but only applies to the next height
  288. lastHeightValsChanged = header.Height + 1
  289. }
  290. // Update validator accums and set state variables
  291. nextValSet.IncrementAccum(1)
  292. // update the params with the latest abciResponses
  293. nextParams := s.ConsensusParams
  294. lastHeightParamsChanged := s.LastHeightConsensusParamsChanged
  295. if abciResponses.EndBlock.ConsensusParamUpdates != nil {
  296. // NOTE: must not mutate s.ConsensusParams
  297. nextParams = s.ConsensusParams.Update(abciResponses.EndBlock.ConsensusParamUpdates)
  298. err := nextParams.Validate()
  299. if err != nil {
  300. return s, fmt.Errorf("Error updating consensus params: %v", err)
  301. }
  302. // change results from this height but only applies to the next height
  303. lastHeightParamsChanged = header.Height + 1
  304. }
  305. // NOTE: the AppHash has not been populated.
  306. // It will be filled on state.Save.
  307. return State{
  308. ChainID: s.ChainID,
  309. LastBlockHeight: header.Height,
  310. LastBlockTotalTx: s.LastBlockTotalTx + header.NumTxs,
  311. LastBlockID: blockID,
  312. LastBlockTime: header.Time,
  313. Validators: nextValSet,
  314. LastValidators: s.Validators.Copy(),
  315. LastHeightValidatorsChanged: lastHeightValsChanged,
  316. ConsensusParams: nextParams,
  317. LastHeightConsensusParamsChanged: lastHeightParamsChanged,
  318. LastResultsHash: abciResponses.ResultsHash(),
  319. AppHash: nil,
  320. }, nil
  321. }
  322. // Fire NewBlock, NewBlockHeader.
  323. // Fire TxEvent for every tx.
  324. // NOTE: if Tendermint crashes before commit, some or all of these events may be published again.
  325. func fireEvents(logger log.Logger, eventBus types.BlockEventPublisher, block *types.Block, abciResponses *ABCIResponses) {
  326. // NOTE: do we still need this buffer ?
  327. txEventBuffer := types.NewTxEventBuffer(eventBus, int(block.NumTxs))
  328. for i, tx := range block.Data.Txs {
  329. txEventBuffer.PublishEventTx(types.EventDataTx{types.TxResult{
  330. Height: block.Height,
  331. Index: uint32(i),
  332. Tx: tx,
  333. Result: *(abciResponses.DeliverTx[i]),
  334. }})
  335. }
  336. eventBus.PublishEventNewBlock(types.EventDataNewBlock{block})
  337. eventBus.PublishEventNewBlockHeader(types.EventDataNewBlockHeader{block.Header})
  338. err := txEventBuffer.Flush()
  339. if err != nil {
  340. logger.Error("Failed to flush event buffer", "err", err)
  341. }
  342. }
  343. //----------------------------------------------------------------------------------------------------
  344. // Execute block without state. TODO: eliminate
  345. // ExecCommitBlock executes and commits a block on the proxyApp without validating or mutating the state.
  346. // It returns the application root hash (result of abci.Commit).
  347. func ExecCommitBlock(appConnConsensus proxy.AppConnConsensus, block *types.Block, logger log.Logger) ([]byte, error) {
  348. _, err := execBlockOnProxyApp(logger, appConnConsensus, block)
  349. if err != nil {
  350. logger.Error("Error executing block on proxy app", "height", block.Height, "err", err)
  351. return nil, err
  352. }
  353. // Commit block, get hash back
  354. res, err := appConnConsensus.CommitSync()
  355. if err != nil {
  356. logger.Error("Client error during proxyAppConn.CommitSync", "err", res)
  357. return nil, err
  358. }
  359. // ResponseCommit has no error or log, just data
  360. return res.Data, nil
  361. }