state: re-order funcs. fix tests

This commit is contained in:
Ethan Buchman
2017-12-27 20:03:48 -05:00
parent 9e6d088757
commit f82b7e2a13
7 changed files with 479 additions and 461 deletions

View File

@@ -1,7 +1,6 @@
package state
import (
"bytes"
"errors"
"fmt"
@@ -14,27 +13,119 @@ import (
"github.com/tendermint/tmlibs/log"
)
//--------------------------------------------------
// Execute the block
//-----------------------------------------------------------------------------
// BlockExecutor handles block execution and state updates.
// It exposes ApplyBlock(), which validates & executes the block, updates state w/ ABCI responses,
// then commits and updates the mempool atomically, then saves state.
// ValExecBlock executes the block and returns the responses. It does NOT mutate State.
// + validates the block
// + executes block.Txs on the proxyAppConn
func (blockExec *BlockExecutor) ValExecBlock(s State, block *types.Block) (*ABCIResponses, error) {
if err := s.validateBlock(block); err != nil {
return nil, ErrInvalidBlock(err)
// BlockExecutor provides the context and accessories for properly executing a block.
type BlockExecutor struct {
db dbm.DB
logger log.Logger
txEventPublisher types.TxEventPublisher
proxyApp proxy.AppConnConsensus
mempool types.Mempool
evpool types.EvidencePool
}
// NewBlockExecutor returns a new BlockExecutor.
func NewBlockExecutor(db dbm.DB, logger log.Logger,
txEventer types.TxEventPublisher, proxyApp proxy.AppConnConsensus,
mempool types.Mempool, evpool types.EvidencePool) *BlockExecutor {
return &BlockExecutor{
db,
logger,
txEventer,
proxyApp,
mempool,
evpool,
}
}
// ApplyBlock validates the block against the state, executes it against the app,
// commits it, and saves the block and state. It's the only function that needs to be called
// from outside this package to process and commit an entire block.
// It takes a blockID to avoid recomputing the parts hash.
func (blockExec *BlockExecutor) ApplyBlock(s State, blockID types.BlockID, block *types.Block) (State, error) {
if err := validateBlock(s, block); err != nil {
return s, ErrInvalidBlock(err)
}
abciResponses, err := execBlockOnProxyApp(blockExec.logger, blockExec.proxyApp, block)
if err != nil {
// There was some error in proxyApp
// TODO Report error and wait for proxyApp to be available.
return nil, ErrProxyAppConn(err)
return s, ErrProxyAppConn(err)
}
return abciResponses, nil
fireEvents(blockExec.txEventPublisher, block, abciResponses)
fail.Fail() // XXX
// save the results before we commit
saveABCIResponses(blockExec.db, block.Height, abciResponses)
fail.Fail() // XXX
// update the state with the block and responses
s, err = updateState(s, blockID, block.Header, abciResponses)
if err != nil {
return s, fmt.Errorf("Commit failed for application: %v", err)
}
// lock mempool, commit state, update mempoool
appHash, err := blockExec.Commit(block)
if err != nil {
return s, fmt.Errorf("Commit failed for application: %v", err)
}
fail.Fail() // XXX
// save the state and the validators
SaveState(blockExec.db, s, appHash)
return s, nil
}
// Commit locks the mempool, runs the ABCI Commit message, and updates the mempool.
// It returns the result of calling abci.Commit (the AppHash), and an error.
// The Mempool must be locked during commit and update because state is typically reset on Commit and old txs must be replayed
// against committed state before new txs are run in the mempool, lest they be invalid.
func (blockExec *BlockExecutor) Commit(block *types.Block) ([]byte, error) {
blockExec.mempool.Lock()
defer blockExec.mempool.Unlock()
// Commit block, get hash back
res, err := blockExec.proxyApp.CommitSync()
if err != nil {
blockExec.logger.Error("Client error during proxyAppConn.CommitSync", "err", err)
return nil, err
}
if res.IsErr() {
blockExec.logger.Error("Error in proxyAppConn.CommitSync", "err", res)
return nil, res
}
if res.Log != "" {
blockExec.logger.Debug("Commit.Log: " + res.Log)
}
blockExec.logger.Info("Committed state", "height", block.Height, "txs", block.NumTxs, "hash", res.Data)
// Update evpool
blockExec.evpool.MarkEvidenceAsCommitted(block.Evidence.Evidence)
// Update mempool.
if err := blockExec.mempool.Update(block.Height, block.Txs); err != nil {
return nil, err
}
return res.Data, nil
}
//---------------------------------------------------------
// Helper functions for executing blocks and updating state
// Executes block's transactions on proxyAppConn.
// Returns a list of transaction results and updates to the validator set
func execBlockOnProxyApp(logger log.Logger, proxyAppConn proxy.AppConnConsensus, block *types.Block) (*ABCIResponses, error) {
@@ -196,189 +287,62 @@ func changeInVotingPowerMoreOrEqualToOneThird(currentSet *types.ValidatorSet, up
return false, nil
}
//-----------------------------------------------------
// Validate block
// updateState returns a new State updated according to the header and responses.
func updateState(s State, blockID types.BlockID, header *types.Header,
abciResponses *ABCIResponses) (State, error) {
// MakeBlock builds a block with the given txs and commit from the current state.
func (s State) MakeBlock(height int64, txs []types.Tx, commit *types.Commit) (*types.Block, *types.PartSet) {
// build base block
block := types.MakeBlock(height, txs, commit)
// copy the valset so we can apply changes from EndBlock
// and update s.LastValidators and s.Validators
prevValSet := s.Validators.Copy()
nextValSet := prevValSet.Copy()
// fill header with state data
block.ChainID = s.ChainID
block.TotalTxs = s.LastBlockTotalTx + block.NumTxs
block.LastBlockID = s.LastBlockID
block.ValidatorsHash = s.Validators.Hash()
block.AppHash = s.AppHash
block.ConsensusHash = s.ConsensusParams.Hash()
block.LastResultsHash = s.LastResultsHash
return block, block.MakePartSet(s.ConsensusParams.BlockGossip.BlockPartSizeBytes)
}
// ValidateBlock validates the block against the state.
func (s State) ValidateBlock(block *types.Block) error {
return s.validateBlock(block)
}
func (s State) validateBlock(b *types.Block) error {
// validate internal consistency
if err := b.ValidateBasic(); err != nil {
return err
}
// validate basic info
if b.ChainID != s.ChainID {
return fmt.Errorf("Wrong Block.Header.ChainID. Expected %v, got %v", s.ChainID, b.ChainID)
}
if b.Height != s.LastBlockHeight+1 {
return fmt.Errorf("Wrong Block.Header.Height. Expected %v, got %v", s.LastBlockHeight+1, b.Height)
}
/* TODO: Determine bounds for Time
See blockchain/reactor "stopSyncingDurationMinutes"
if !b.Time.After(lastBlockTime) {
return errors.New("Invalid Block.Header.Time")
}
*/
// validate prev block info
if !b.LastBlockID.Equals(s.LastBlockID) {
return fmt.Errorf("Wrong Block.Header.LastBlockID. Expected %v, got %v", s.LastBlockID, b.LastBlockID)
}
newTxs := int64(len(b.Data.Txs))
if b.TotalTxs != s.LastBlockTotalTx+newTxs {
return fmt.Errorf("Wrong Block.Header.TotalTxs. Expected %v, got %v", s.LastBlockTotalTx+newTxs, b.TotalTxs)
}
// validate app info
if !bytes.Equal(b.AppHash, s.AppHash) {
return fmt.Errorf("Wrong Block.Header.AppHash. Expected %X, got %v", s.AppHash, b.AppHash)
}
if !bytes.Equal(b.ConsensusHash, s.ConsensusParams.Hash()) {
return fmt.Errorf("Wrong Block.Header.ConsensusHash. Expected %X, got %v", s.ConsensusParams.Hash(), b.ConsensusHash)
}
if !bytes.Equal(b.LastResultsHash, s.LastResultsHash) {
return fmt.Errorf("Wrong Block.Header.LastResultsHash. Expected %X, got %v", s.LastResultsHash, b.LastResultsHash)
}
if !bytes.Equal(b.ValidatorsHash, s.Validators.Hash()) {
return fmt.Errorf("Wrong Block.Header.ValidatorsHash. Expected %X, got %v", s.Validators.Hash(), b.ValidatorsHash)
}
// Validate block LastCommit.
if b.Height == 1 {
if len(b.LastCommit.Precommits) != 0 {
return errors.New("Block at height 1 (first block) should have no LastCommit precommits")
}
} else {
if len(b.LastCommit.Precommits) != s.LastValidators.Size() {
return fmt.Errorf("Invalid block commit size. Expected %v, got %v",
s.LastValidators.Size(), len(b.LastCommit.Precommits))
}
err := s.LastValidators.VerifyCommit(
s.ChainID, s.LastBlockID, b.Height-1, b.LastCommit)
// update the validator set with the latest abciResponses
lastHeightValsChanged := s.LastHeightValidatorsChanged
if len(abciResponses.EndBlock.ValidatorUpdates) > 0 {
err := updateValidators(nextValSet, abciResponses.EndBlock.ValidatorUpdates)
if err != nil {
return err
return s, fmt.Errorf("Error changing validator set: %v", err)
}
// change results from this height but only applies to the next height
lastHeightValsChanged = header.Height + 1
}
for _, ev := range b.Evidence.Evidence {
if err := VerifyEvidence(s, ev); err != nil {
return types.NewEvidenceInvalidErr(ev, err)
}
/* // Needs a db ...
valset, err := LoadValidators(s.db, ev.Height())
// Update validator accums and set state variables
nextValSet.IncrementAccum(1)
// update the params with the latest abciResponses
nextParams := s.ConsensusParams
lastHeightParamsChanged := s.LastHeightConsensusParamsChanged
if abciResponses.EndBlock.ConsensusParamUpdates != nil {
// NOTE: must not mutate s.ConsensusParams
nextParams = s.ConsensusParams.Update(abciResponses.EndBlock.ConsensusParamUpdates)
err := nextParams.Validate()
if err != nil {
// XXX/TODO: what do we do if we can't load the valset?
// eg. if we have pruned the state or height is too high?
return err
return s, fmt.Errorf("Error updating consensus params: %v", err)
}
if err := VerifyEvidenceValidator(valSet, ev); err != nil {
return types.NewEvidenceInvalidErr(ev, err)
}
*/
// change results from this height but only applies to the next height
lastHeightParamsChanged = header.Height + 1
}
return nil
// NOTE: the AppHash has not been populated.
// It will be filled on state.Save.
return State{
ChainID: s.ChainID,
LastBlockHeight: header.Height,
LastBlockTotalTx: s.LastBlockTotalTx + header.NumTxs,
LastBlockID: blockID,
LastBlockTime: header.Time,
Validators: nextValSet,
LastValidators: s.Validators.Copy(),
LastHeightValidatorsChanged: lastHeightValsChanged,
ConsensusParams: nextParams,
LastHeightConsensusParamsChanged: lastHeightParamsChanged,
LastResultsHash: abciResponses.ResultsHash(),
AppHash: nil,
}, nil
}
// XXX: What's cheaper (ie. what should be checked first):
// evidence internal validity (ie. sig checks) or validator existed (fetch historical val set from db)
// VerifyEvidence verifies the evidence fully by checking it is internally
// consistent and sufficiently recent.
func VerifyEvidence(s State, evidence types.Evidence) error {
height := s.LastBlockHeight
evidenceAge := height - evidence.Height()
maxAge := s.ConsensusParams.EvidenceParams.MaxAge
if evidenceAge > maxAge {
return fmt.Errorf("Evidence from height %d is too old. Min height is %d",
evidence.Height(), height-maxAge)
}
if err := evidence.Verify(s.ChainID); err != nil {
return err
}
return nil
}
// VerifyEvidenceValidator returns the voting power of the validator at the height of the evidence.
// It returns an error if the validator did not exist or does not match that loaded from the historical validator set.
func VerifyEvidenceValidator(valset *types.ValidatorSet, evidence types.Evidence) (priority int64, err error) {
// The address must have been an active validator at the height
ev := evidence
height, addr, idx := ev.Height(), ev.Address(), ev.Index()
valIdx, val := valset.GetByAddress(addr)
if val == nil {
return priority, fmt.Errorf("Address %X was not a validator at height %d", addr, height)
} else if idx != valIdx {
return priority, fmt.Errorf("Address %X was validator %d at height %d, not %d", addr, valIdx, height, idx)
}
priority = val.VotingPower
return priority, nil
}
//-----------------------------------------------------------------------------
// ApplyBlock validates & executes the block, updates state w/ ABCI responses,
// then commits and updates the mempool atomically, then saves state.
// BlockExecutor provides the context and accessories for properly executing a block.
type BlockExecutor struct {
db dbm.DB
logger log.Logger
txEventPublisher types.TxEventPublisher
proxyApp proxy.AppConnConsensus
mempool types.Mempool
evpool types.EvidencePool
}
func NewBlockExecutor(db dbm.DB, logger log.Logger, txEventer types.TxEventPublisher, proxyApp proxy.AppConnConsensus,
mempool types.Mempool, evpool types.EvidencePool) *BlockExecutor {
return &BlockExecutor{
db,
logger,
txEventer,
proxyApp,
mempool,
evpool,
}
}
// ApplyBlock validates the block against the state, executes it against the app,
// commits it, and saves the block and state. It's the only function that needs to be called
// from outside this package to process and commit an entire block.
// It takes a blockID to avoid recomputing the parts hash.
func (blockExec *BlockExecutor) ApplyBlock(s State, blockID types.BlockID, block *types.Block) (State, error) {
abciResponses, err := blockExec.ValExecBlock(s, block)
if err != nil {
return s, fmt.Errorf("Exec failed for application: %v", err)
}
func fireEvents(txEventPublisher types.TxEventPublisher, block *types.Block, abciResponses *ABCIResponses) {
// TODO: Fire events
/*
tx := types.Tx(req.GetDeliverTx().Tx)
@@ -389,68 +353,10 @@ func (blockExec *BlockExecutor) ApplyBlock(s State, blockID types.BlockID, block
Result: *txRes,
}})
*/
fail.Fail() // XXX
// save the results before we commit
SaveABCIResponses(blockExec.db, block.Height, abciResponses)
fail.Fail() // XXX
// update the state with the block and responses
s, err = s.NextState(blockID, block.Header, abciResponses)
if err != nil {
return s, fmt.Errorf("Commit failed for application: %v", err)
}
// lock mempool, commit state, update mempoool
appHash, err := blockExec.Commit(block)
if err != nil {
return s, fmt.Errorf("Commit failed for application: %v", err)
}
fail.Fail() // XXX
// save the state and the validators
s.Save(blockExec.db, appHash)
return s, nil
}
// Commit locks the mempool, runs the ABCI Commit message, and updates the mempool.
// It returns the result of calling abci.Commit (the AppHash), and an error.
// The Mempool must be locked during commit and update because state is typically reset on Commit and old txs must be replayed
// against committed state before new txs are run in the mempool, lest they be invalid.
func (blockExec *BlockExecutor) Commit(block *types.Block) ([]byte, error) {
blockExec.mempool.Lock()
defer blockExec.mempool.Unlock()
// Commit block, get hash back
res, err := blockExec.proxyApp.CommitSync()
if err != nil {
blockExec.logger.Error("Client error during proxyAppConn.CommitSync", "err", err)
return nil, err
}
if res.IsErr() {
blockExec.logger.Error("Error in proxyAppConn.CommitSync", "err", res)
return nil, res
}
if res.Log != "" {
blockExec.logger.Debug("Commit.Log: " + res.Log)
}
blockExec.logger.Info("Committed state", "height", block.Height, "txs", block.NumTxs, "hash", res.Data)
// Update evpool
blockExec.evpool.MarkEvidenceAsCommitted(block.Evidence.Evidence)
// Update mempool.
if err := blockExec.mempool.Update(block.Height, block.Txs); err != nil {
return nil, err
}
return res.Data, nil
}
//----------------------------------------------------------------------------------------------------
// Execute block without state. TODO: eliminate
// ExecCommitBlock executes and commits a block on the proxyApp without validating or mutating the state.
// It returns the application root hash (result of abci.Commit).