package state import ( "bytes" "errors" "fmt" fail "github.com/ebuchman/fail-test" abci "github.com/tendermint/abci/types" crypto "github.com/tendermint/go-crypto" "github.com/tendermint/tendermint/proxy" "github.com/tendermint/tendermint/types" dbm "github.com/tendermint/tmlibs/db" "github.com/tendermint/tmlibs/log" ) //-------------------------------------------------- // Execute the block // 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) } 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 abciResponses, nil } // 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) { var validTxs, invalidTxs = 0, 0 txIndex := 0 abciResponses := NewABCIResponses(block) // Execute transactions and get hash proxyCb := func(req *abci.Request, res *abci.Response) { switch r := res.Value.(type) { case *abci.Response_DeliverTx: // TODO: make use of res.Log // TODO: make use of this info // Blocks may include invalid txs. txRes := r.DeliverTx if txRes.Code == abci.CodeTypeOK { validTxs++ } else { logger.Debug("Invalid tx", "code", txRes.Code, "log", txRes.Log) invalidTxs++ } abciResponses.DeliverTx[txIndex] = txRes txIndex++ } } proxyAppConn.SetResponseCallback(proxyCb) // determine which validators did not sign last block absentVals := make([]int32, 0) for valI, vote := range block.LastCommit.Precommits { if vote == nil { absentVals = append(absentVals, int32(valI)) } } // TODO: determine which validators were byzantine // Begin block _, err := proxyAppConn.BeginBlockSync(abci.RequestBeginBlock{ Hash: block.Hash(), Header: types.TM2PB.Header(block.Header), AbsentValidators: absentVals, ByzantineValidators: nil, }) if err != nil { logger.Error("Error in proxyAppConn.BeginBlock", "err", err) return nil, err } // Run txs of block for _, tx := range block.Txs { proxyAppConn.DeliverTxAsync(tx) if err := proxyAppConn.Error(); err != nil { return nil, err } } // End block abciResponses.EndBlock, err = proxyAppConn.EndBlockSync(abci.RequestEndBlock{block.Height}) if err != nil { logger.Error("Error in proxyAppConn.EndBlock", "err", err) return nil, err } logger.Info("Executed block", "height", block.Height, "validTxs", validTxs, "invalidTxs", invalidTxs) valUpdates := abciResponses.EndBlock.ValidatorUpdates if len(valUpdates) > 0 { logger.Info("Updates to validators", "updates", abci.ValidatorsString(valUpdates)) } return abciResponses, nil } func updateValidators(currentSet *types.ValidatorSet, updates []*abci.Validator) error { // If more or equal than 1/3 of total voting power changed in one block, then // a light client could never prove the transition externally. See // ./lite/doc.go for details on how a light client tracks validators. vp23, err := changeInVotingPowerMoreOrEqualToOneThird(currentSet, updates) if err != nil { return err } if vp23 { return errors.New("the change in voting power must be strictly less than 1/3") } for _, v := range updates { pubkey, err := crypto.PubKeyFromBytes(v.PubKey) // NOTE: expects go-wire encoded pubkey if err != nil { return err } address := pubkey.Address() power := int64(v.Power) // mind the overflow from int64 if power < 0 { return fmt.Errorf("Power (%d) overflows int64", v.Power) } _, val := currentSet.GetByAddress(address) if val == nil { // add val added := currentSet.Add(types.NewValidator(pubkey, power)) if !added { return fmt.Errorf("Failed to add new validator %X with voting power %d", address, power) } } else if v.Power == 0 { // remove val _, removed := currentSet.Remove(address) if !removed { return fmt.Errorf("Failed to remove validator %X", address) } } else { // update val val.VotingPower = power updated := currentSet.Update(val) if !updated { return fmt.Errorf("Failed to update validator %X with voting power %d", address, power) } } } return nil } func changeInVotingPowerMoreOrEqualToOneThird(currentSet *types.ValidatorSet, updates []*abci.Validator) (bool, error) { threshold := currentSet.TotalVotingPower() * 1 / 3 acc := int64(0) for _, v := range updates { pubkey, err := crypto.PubKeyFromBytes(v.PubKey) // NOTE: expects go-wire encoded pubkey if err != nil { return false, err } address := pubkey.Address() power := int64(v.Power) // mind the overflow from int64 if power < 0 { return false, fmt.Errorf("Power (%d) overflows int64", v.Power) } _, val := currentSet.GetByAddress(address) if val == nil { acc += power } else { np := val.VotingPower - power if np < 0 { np = -np } acc += np } if acc >= threshold { return true, nil } } return false, nil } //----------------------------------------------------- // Validate block // 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) // 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) if err != nil { return err } } 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()) 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 } if err := VerifyEvidenceValidator(valSet, ev); err != nil { return types.NewEvidenceInvalidErr(ev, err) } */ } return 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) } // TODO: Fire events /* tx := types.Tx(req.GetDeliverTx().Tx) txEventPublisher.PublishEventTx(types.EventDataTx{types.TxResult{ Height: block.Height, Index: uint32(txIndex), Tx: tx, 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 } // 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). func ExecCommitBlock(appConnConsensus proxy.AppConnConsensus, block *types.Block, logger log.Logger) ([]byte, error) { _, err := execBlockOnProxyApp(logger, appConnConsensus, block) if err != nil { logger.Error("Error executing block on proxy app", "height", block.Height, "err", err) return nil, err } // Commit block, get hash back res, err := appConnConsensus.CommitSync() if err != nil { logger.Error("Client error during proxyAppConn.CommitSync", "err", res) return nil, err } if res.IsErr() { logger.Error("Error in proxyAppConn.CommitSync", "err", res) return nil, res } if res.Log != "" { logger.Info("Commit.Log: " + res.Log) } return res.Data, nil }