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.

306 lines
8.9 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
  1. package state
  2. import (
  3. "errors"
  4. "fmt"
  5. fail "github.com/ebuchman/fail-test"
  6. abci "github.com/tendermint/abci/types"
  7. . "github.com/tendermint/go-common"
  8. crypto "github.com/tendermint/go-crypto"
  9. "github.com/tendermint/tendermint/proxy"
  10. "github.com/tendermint/tendermint/state/txindex"
  11. "github.com/tendermint/tendermint/types"
  12. )
  13. //--------------------------------------------------
  14. // Execute the block
  15. // ValExecBlock executes the block, but does NOT mutate State.
  16. // + validates the block
  17. // + executes block.Txs on the proxyAppConn
  18. func (s *State) ValExecBlock(eventCache types.Fireable, proxyAppConn proxy.AppConnConsensus, block *types.Block) (*ABCIResponses, error) {
  19. // Validate the block.
  20. if err := s.validateBlock(block); err != nil {
  21. return nil, ErrInvalidBlock(err)
  22. }
  23. // Execute the block txs
  24. abciResponses, err := execBlockOnProxyApp(eventCache, proxyAppConn, block)
  25. if err != nil {
  26. // There was some error in proxyApp
  27. // TODO Report error and wait for proxyApp to be available.
  28. return nil, ErrProxyAppConn(err)
  29. }
  30. return abciResponses, nil
  31. }
  32. // Executes block's transactions on proxyAppConn.
  33. // Returns a list of transaction results and updates to the validator set
  34. // TODO: Generate a bitmap or otherwise store tx validity in state.
  35. func execBlockOnProxyApp(eventCache types.Fireable, proxyAppConn proxy.AppConnConsensus, block *types.Block) (*ABCIResponses, error) {
  36. var validTxs, invalidTxs = 0, 0
  37. txIndex := 0
  38. abciResponses := NewABCIResponses(block)
  39. // Execute transactions and get hash
  40. proxyCb := func(req *abci.Request, res *abci.Response) {
  41. switch r := res.Value.(type) {
  42. case *abci.Response_DeliverTx:
  43. // TODO: make use of res.Log
  44. // TODO: make use of this info
  45. // Blocks may include invalid txs.
  46. // reqDeliverTx := req.(abci.RequestDeliverTx)
  47. txError := ""
  48. txResult := r.DeliverTx
  49. if txResult.Code == abci.CodeType_OK {
  50. validTxs++
  51. } else {
  52. log.Debug("Invalid tx", "code", txResult.Code, "log", txResult.Log)
  53. invalidTxs++
  54. txError = txResult.Code.String()
  55. }
  56. abciResponses.DeliverTx[txIndex] = txResult
  57. txIndex++
  58. // NOTE: if we count we can access the tx from the block instead of
  59. // pulling it from the req
  60. event := types.EventDataTx{
  61. Height: block.Height,
  62. Tx: types.Tx(req.GetDeliverTx().Tx),
  63. Data: txResult.Data,
  64. Code: txResult.Code,
  65. Log: txResult.Log,
  66. Error: txError,
  67. }
  68. types.FireEventTx(eventCache, event)
  69. }
  70. }
  71. proxyAppConn.SetResponseCallback(proxyCb)
  72. // Begin block
  73. err := proxyAppConn.BeginBlockSync(block.Hash(), types.TM2PB.Header(block.Header))
  74. if err != nil {
  75. log.Warn("Error in proxyAppConn.BeginBlock", "error", err)
  76. return nil, err
  77. }
  78. // Run txs of block
  79. for _, tx := range block.Txs {
  80. proxyAppConn.DeliverTxAsync(tx)
  81. if err := proxyAppConn.Error(); err != nil {
  82. return nil, err
  83. }
  84. }
  85. // End block
  86. abciResponses.EndBlock, err = proxyAppConn.EndBlockSync(uint64(block.Height))
  87. if err != nil {
  88. log.Warn("Error in proxyAppConn.EndBlock", "error", err)
  89. return nil, err
  90. }
  91. valDiff := abciResponses.EndBlock.Diffs
  92. log.Info("Executed block", "height", block.Height, "valid txs", validTxs, "invalid txs", invalidTxs)
  93. if len(valDiff) > 0 {
  94. log.Info("Update to validator set", "updates", abci.ValidatorsString(valDiff))
  95. }
  96. return abciResponses, nil
  97. }
  98. func updateValidators(validators *types.ValidatorSet, changedValidators []*abci.Validator) error {
  99. // TODO: prevent change of 1/3+ at once
  100. for _, v := range changedValidators {
  101. pubkey, err := crypto.PubKeyFromBytes(v.PubKey) // NOTE: expects go-wire encoded pubkey
  102. if err != nil {
  103. return err
  104. }
  105. address := pubkey.Address()
  106. power := int64(v.Power)
  107. // mind the overflow from uint64
  108. if power < 0 {
  109. return errors.New(Fmt("Power (%d) overflows int64", v.Power))
  110. }
  111. _, val := validators.GetByAddress(address)
  112. if val == nil {
  113. // add val
  114. added := validators.Add(types.NewValidator(pubkey, power))
  115. if !added {
  116. return errors.New(Fmt("Failed to add new validator %X with voting power %d", address, power))
  117. }
  118. } else if v.Power == 0 {
  119. // remove val
  120. _, removed := validators.Remove(address)
  121. if !removed {
  122. return errors.New(Fmt("Failed to remove validator %X)"))
  123. }
  124. } else {
  125. // update val
  126. val.VotingPower = power
  127. updated := validators.Update(val)
  128. if !updated {
  129. return errors.New(Fmt("Failed to update validator %X with voting power %d", address, power))
  130. }
  131. }
  132. }
  133. return nil
  134. }
  135. // return a bit array of validators that signed the last commit
  136. // NOTE: assumes commits have already been authenticated
  137. func commitBitArrayFromBlock(block *types.Block) *BitArray {
  138. signed := NewBitArray(len(block.LastCommit.Precommits))
  139. for i, precommit := range block.LastCommit.Precommits {
  140. if precommit != nil {
  141. signed.SetIndex(i, true) // val_.LastCommitHeight = block.Height - 1
  142. }
  143. }
  144. return signed
  145. }
  146. //-----------------------------------------------------
  147. // Validate block
  148. func (s *State) ValidateBlock(block *types.Block) error {
  149. return s.validateBlock(block)
  150. }
  151. func (s *State) validateBlock(block *types.Block) error {
  152. // Basic block validation.
  153. err := block.ValidateBasic(s.ChainID, s.LastBlockHeight, s.LastBlockID, s.LastBlockTime, s.AppHash)
  154. if err != nil {
  155. return err
  156. }
  157. // Validate block LastCommit.
  158. if block.Height == 1 {
  159. if len(block.LastCommit.Precommits) != 0 {
  160. return errors.New("Block at height 1 (first block) should have no LastCommit precommits")
  161. }
  162. } else {
  163. if len(block.LastCommit.Precommits) != s.LastValidators.Size() {
  164. return errors.New(Fmt("Invalid block commit size. Expected %v, got %v",
  165. s.LastValidators.Size(), len(block.LastCommit.Precommits)))
  166. }
  167. err := s.LastValidators.VerifyCommit(
  168. s.ChainID, s.LastBlockID, block.Height-1, block.LastCommit)
  169. if err != nil {
  170. return err
  171. }
  172. }
  173. return nil
  174. }
  175. //-----------------------------------------------------------------------------
  176. // ApplyBlock validates & executes the block, updates state w/ ABCI responses,
  177. // then commits and updates the mempool atomically, then saves state.
  178. // Transaction results are optionally indexed.
  179. // Validate, execute, and commit block against app, save block and state
  180. func (s *State) ApplyBlock(eventCache types.Fireable, proxyAppConn proxy.AppConnConsensus,
  181. block *types.Block, partsHeader types.PartSetHeader, mempool types.Mempool) error {
  182. abciResponses, err := s.ValExecBlock(eventCache, proxyAppConn, block)
  183. if err != nil {
  184. return fmt.Errorf("Exec failed for application: %v", err)
  185. }
  186. fail.Fail() // XXX
  187. // index txs. This could run in the background
  188. s.indexTxs(abciResponses)
  189. // save the results before we commit
  190. s.SaveABCIResponses(abciResponses)
  191. fail.Fail() // XXX
  192. // now update the block and validators
  193. s.SetBlockAndValidators(block.Header, partsHeader, abciResponses)
  194. // lock mempool, commit state, update mempoool
  195. err = s.CommitStateUpdateMempool(proxyAppConn, block, mempool)
  196. if err != nil {
  197. return fmt.Errorf("Commit failed for application: %v", err)
  198. }
  199. fail.Fail() // XXX
  200. // save the state
  201. s.Save()
  202. return nil
  203. }
  204. // mempool must be locked during commit and update
  205. // because state is typically reset on Commit and old txs must be replayed
  206. // against committed state before new txs are run in the mempool, lest they be invalid
  207. func (s *State) CommitStateUpdateMempool(proxyAppConn proxy.AppConnConsensus, block *types.Block, mempool types.Mempool) error {
  208. mempool.Lock()
  209. defer mempool.Unlock()
  210. // Commit block, get hash back
  211. res := proxyAppConn.CommitSync()
  212. if res.IsErr() {
  213. log.Warn("Error in proxyAppConn.CommitSync", "error", res)
  214. return res
  215. }
  216. if res.Log != "" {
  217. log.Debug("Commit.Log: " + res.Log)
  218. }
  219. log.Info("Committed state", "hash", res.Data)
  220. // Set the state's new AppHash
  221. s.AppHash = res.Data
  222. // Update mempool.
  223. mempool.Update(block.Height, block.Txs)
  224. return nil
  225. }
  226. func (s *State) indexTxs(abciResponses *ABCIResponses) {
  227. // save the tx results using the TxIndexer
  228. // NOTE: these may be overwriting, but the values should be the same.
  229. batch := txindex.NewBatch(len(abciResponses.DeliverTx))
  230. for i, d := range abciResponses.DeliverTx {
  231. tx := abciResponses.txs[i]
  232. batch.Add(types.TxResult{
  233. Height: uint64(abciResponses.Height),
  234. Index: uint32(i),
  235. Tx: tx,
  236. Result: *d,
  237. })
  238. }
  239. s.TxIndexer.AddBatch(batch)
  240. }
  241. // Exec and commit a block on the proxyApp without validating or mutating the state
  242. // Returns the application root hash (result of abci.Commit)
  243. func ExecCommitBlock(appConnConsensus proxy.AppConnConsensus, block *types.Block) ([]byte, error) {
  244. var eventCache types.Fireable // nil
  245. _, err := execBlockOnProxyApp(eventCache, appConnConsensus, block)
  246. if err != nil {
  247. log.Warn("Error executing block on proxy app", "height", block.Height, "err", err)
  248. return nil, err
  249. }
  250. // Commit block, get hash back
  251. res := appConnConsensus.CommitSync()
  252. if res.IsErr() {
  253. log.Warn("Error in proxyAppConn.CommitSync", "error", res)
  254. return nil, res
  255. }
  256. if res.Log != "" {
  257. log.Info("Commit.Log: " + res.Log)
  258. }
  259. return res.Data, nil
  260. }