From 2dfcbcc310f7dfab1078f6bd2386c1e5bb159412 Mon Sep 17 00:00:00 2001 From: Callum Waters Date: Thu, 29 Sep 2022 09:58:23 +0200 Subject: [PATCH] clean up kvstore example --- abci/cmd/abci-cli/abci-cli.go | 6 +- abci/example/kvstore/code.go | 11 +- abci/example/kvstore/helpers.go | 34 ++ abci/example/kvstore/kvstore.go | 505 ++++++++++++++------------- abci/example/kvstore/kvstore_test.go | 142 ++++---- consensus/byzantine_test.go | 2 +- consensus/common_test.go | 1 + consensus/mempool_test.go | 33 +- types/events.go | 2 +- 9 files changed, 394 insertions(+), 342 deletions(-) diff --git a/abci/cmd/abci-cli/abci-cli.go b/abci/cmd/abci-cli/abci-cli.go index 23ba481be..d1d0e4f44 100644 --- a/abci/cmd/abci-cli/abci-cli.go +++ b/abci/cmd/abci-cli/abci-cli.go @@ -319,7 +319,7 @@ func cmdTest(cmd *cobra.Command, args []string) error { return servertest.FinalizeBlock(ctx, client, [][]byte{ []byte("abc"), }, []uint32{ - kvstore.CodeTypeBadNonce, + kvstore.CodeTypeInvalidTxFormat, }, nil, nil) }, func() error { return servertest.Commit(ctx, client) }, @@ -340,12 +340,12 @@ func cmdTest(cmd *cobra.Command, args []string) error { {0x00, 0x00, 0x04}, {0x00, 0x00, 0x06}, }, []uint32{ - kvstore.CodeTypeBadNonce, + kvstore.CodeTypeInvalidTxFormat, kvstore.CodeTypeOK, kvstore.CodeTypeOK, kvstore.CodeTypeOK, kvstore.CodeTypeOK, - kvstore.CodeTypeBadNonce, + kvstore.CodeTypeInvalidTxFormat, }, nil, []byte{0, 0, 0, 0, 0, 0, 0, 5}) }, func() error { return servertest.Commit(ctx, client) }, diff --git a/abci/example/kvstore/code.go b/abci/example/kvstore/code.go index 5a0fed2e8..f58cfb83c 100644 --- a/abci/example/kvstore/code.go +++ b/abci/example/kvstore/code.go @@ -2,10 +2,9 @@ package kvstore // Return codes for the examples const ( - CodeTypeOK uint32 = 0 - CodeTypeEncodingError uint32 = 1 - CodeTypeBadNonce uint32 = 2 - CodeTypeUnauthorized uint32 = 3 - CodeTypeUnknownError uint32 = 4 - CodeTypeExecuted uint32 = 5 + CodeTypeOK uint32 = 0 + CodeTypeEncodingError uint32 = 1 + CodeTypeInvalidTxFormat uint32 = 2 + CodeTypeUnauthorized uint32 = 3 + CodeTypeExecuted uint32 = 5 ) diff --git a/abci/example/kvstore/helpers.go b/abci/example/kvstore/helpers.go index b70915c92..91c68ca5a 100644 --- a/abci/example/kvstore/helpers.go +++ b/abci/example/kvstore/helpers.go @@ -2,9 +2,15 @@ package kvstore import ( "context" + "encoding/base64" + "fmt" + "strings" "github.com/tendermint/tendermint/abci/types" + cryptoencoding "github.com/tendermint/tendermint/crypto/encoding" + "github.com/tendermint/tendermint/libs/rand" tmrand "github.com/tendermint/tendermint/libs/rand" + "github.com/tendermint/tendermint/proto/tendermint/crypto" ) // RandVal creates one random validator, with a key derived @@ -37,3 +43,31 @@ func InitKVStore(ctx context.Context, app *Application) error { }) return err } + +// Create a new transaction +func NewTx(key, value string) []byte { + return []byte(strings.Join([]string{key, value}, "=")) +} + +func NewRandomTx() []byte { + return NewTx(rand.Str(5), rand.Str(10)) +} + +func NewRandomTxs(n int) [][]byte { + txs := make([][]byte, n) + for i := 0; i < n; i++ { + txs[i] = NewRandomTx() + } + return txs +} + +// Create a transaction to add/remove/update a validator +// To remove, set power to 0. +func MakeValSetChangeTx(pubkey crypto.PublicKey, power int64) []byte { + pk, err := cryptoencoding.PubKeyFromProto(pubkey) + if err != nil { + panic(err) + } + pubStr := base64.StdEncoding.EncodeToString(pk.Bytes()) + return []byte(fmt.Sprintf("%s%s!%d", ValidatorPrefix, pubStr, power)) +} diff --git a/abci/example/kvstore/kvstore.go b/abci/example/kvstore/kvstore.go index c1f12ff36..58f891e84 100644 --- a/abci/example/kvstore/kvstore.go +++ b/abci/example/kvstore/kvstore.go @@ -22,66 +22,32 @@ import ( var ( stateKey = []byte("stateKey") kvPairPrefixKey = []byte("kvPairKey:") - - AppVersion uint64 = 0x1 ) -type State struct { - db dbm.DB - Size int64 `json:"size"` - Height int64 `json:"height"` - AppHash []byte `json:"app_hash"` -} - -func loadState(db dbm.DB) State { - var state State - state.db = db - stateBytes, err := db.Get(stateKey) - if err != nil { - panic(err) - } - if len(stateBytes) == 0 { - return state - } - err = json.Unmarshal(stateBytes, &state) - if err != nil { - panic(err) - } - return state -} - -func saveState(state State) { - stateBytes, err := json.Marshal(state) - if err != nil { - panic(err) - } - err = state.db.Set(stateKey, stateBytes) - if err != nil { - panic(err) - } -} - -func prefixKey(key []byte) []byte { - return append(kvPairPrefixKey, key...) -} - -//--------------------------------------------------- +const ( + ValidatorPrefix = "val=" + AppVersion uint64 = 1 +) var _ types.Application = (*Application)(nil) +// Application is the kvstore state machine. It complies with the abci.Application interface. +// It takes transactions in the form of key=value and saves them in a database. This is +// a somewhat trivial example as there is no real state execution type Application struct { types.BaseApplication state State RetainBlocks int64 // blocks to retain after commit (via ResponseCommit.RetainHeight) - txToRemove map[string]struct{} + stagedTxs [][]byte logger log.Logger // validator set - ValUpdates []types.ValidatorUpdate + valUpdates []types.ValidatorUpdate valAddrToPubKeyMap map[string]cryptoproto.PublicKey } +// NewApplication creates an instance of the kvstore from the provided database func NewApplication(db dbm.DB) *Application { return &Application{ logger: log.NewNopLogger(), @@ -90,6 +56,7 @@ func NewApplication(db dbm.DB) *Application { } } +// NewPersistentApplication creates a new application using the goleveldb database engine func NewPersistentApplication(dbDir string) *Application { name := "kvstore" db, err := dbm.NewGoLevelDB(name, dbDir) @@ -99,81 +66,137 @@ func NewPersistentApplication(dbDir string) *Application { return NewApplication(db) } +// NewInMemoryApplication creates a new application from an in memory database. +// Nothing will be persisted. func NewInMemoryApplication() *Application { return NewApplication(dbm.NewMemDB()) } -func (app *Application) InitChain(_ context.Context, req *types.RequestInitChain) (*types.ResponseInitChain, error) { - for _, v := range req.Validators { - r := app.updateValidator(v) - if r.IsErr() { - app.logger.Error("error updating validators", "r", r) - panic("problem updating validators") +// Info returns information about the state of the application. This is generally used everytime a Tendermint instance +// begins and let's the application know what Tendermint versions it's interacting with. Based from this information, +// Tendermint will ensure it is in sync with the application by potentially replaying the blocks it has. If the +// Application returns a 0 appBlockHeight, Tendermint will call InitChain to initialize the application with consensus related data +func (app *Application) Info(_ context.Context, req *types.RequestInfo) (*types.ResponseInfo, error) { + appHash := make([]byte, 8) + binary.PutVarint(appHash, app.state.Size) + + // Tendermint expects the application to persist validators, on start-up we need to reload them to memory if they exist + if len(app.valAddrToPubKeyMap) == 0 && app.state.Height > 0 { + validators := app.getValidators() + for _, v := range validators { + pubkey, err := cryptoencoding.PubKeyFromProto(v.PubKey) + if err != nil { + panic(fmt.Errorf("can't decode public key: %w", err)) + } + app.valAddrToPubKeyMap[string(pubkey.Address())] = v.PubKey } } - return &types.ResponseInitChain{}, nil -} -func (app *Application) Info(_ context.Context, req *types.RequestInfo) (*types.ResponseInfo, error) { return &types.ResponseInfo{ Data: fmt.Sprintf("{\"size\":%v}", app.state.Size), Version: version.ABCIVersion, AppVersion: AppVersion, LastBlockHeight: app.state.Height, - LastBlockAppHash: app.state.AppHash, + LastBlockAppHash: appHash, }, nil } -// tx is either "val:pubkey!power" or "key=value" or just arbitrary bytes -func (app *Application) handleTx(tx []byte) *types.ExecTxResult { - // if it starts with "val:", update the validator set - // format is "val:pubkey!power" - if isValidatorTx(tx) { - // update validators in the merkle tree - // and in app.ValUpdates - return app.execValidatorTx(tx) +// InitChain takes the genesis validators and stores them in the kvstore. It returns the application hash in the +// case that the application starts prepopulated with values. This method is called whenever a new instance of the application +// starts (i.e. app height = 0). +func (app *Application) InitChain(_ context.Context, req *types.RequestInitChain) (*types.ResponseInitChain, error) { + for _, v := range req.Validators { + app.updateValidator(v) } - - if isPrepareTx(tx) { - return app.execPrepareTx(tx) - } - - var key, value string - parts := bytes.Split(tx, []byte("=")) - if len(parts) == 2 { - key, value = string(parts[0]), string(parts[1]) - } else { - key, value = string(tx), string(tx) - } - - err := app.state.db.Set(prefixKey([]byte(key)), []byte(value)) - if err != nil { - panic(err) - } - app.state.Size++ - - events := []types.Event{ - { - Type: "app", - Attributes: []types.EventAttribute{ - {Key: "creator", Value: "Cosmoshi Netowoko", Index: true}, - {Key: "key", Value: key, Index: true}, - {Key: "index_key", Value: "index is working", Index: true}, - {Key: "noindex_key", Value: "index is working", Index: false}, - }, - }, - } - - return &types.ExecTxResult{Code: CodeTypeOK, Events: events} + appHash := make([]byte, 8) + binary.PutVarint(appHash, app.state.Size) + return &types.ResponseInitChain{ + AppHash: appHash, + }, nil } -func (app *Application) Close() error { - return app.state.db.Close() +// CheckTx handles inbound transactions or in the case of recheckTx assesses old transaction validity after a state transition. +// As this is called frequently, it's preferably to keep the check as stateless and as quick as possible. +// Here we check that the transaction has the correctly key=value format. +func (app *Application) CheckTx(_ context.Context, req *types.RequestCheckTx) (*types.ResponseCheckTx, error) { + if !isValidTx(req.Tx) { + return &types.ResponseCheckTx{Code: CodeTypeInvalidTxFormat}, nil + } + + // If it is a validator update transaction, check that it is correctly formatted + if isValidatorTx(req.Tx) { + if _, _, err := parseValidatorTx(req.Tx); err != nil { + return &types.ResponseCheckTx{Code: CodeTypeInvalidTxFormat}, nil + } + } + + return &types.ResponseCheckTx{Code: CodeTypeOK, GasWanted: 1}, nil } +// Tx must have a format like key:value or key=value. That is: +// - it must have one and only one ":" or "=" +// - It must not begin or end with these special characters +func isValidTx(tx []byte) bool { + if bytes.Count(tx, []byte(":")) == 1 && bytes.Count(tx, []byte("=")) == 0 { + if !bytes.HasPrefix(tx, []byte(":")) && !bytes.HasSuffix(tx, []byte(":")) { + return true + } + } else if bytes.Count(tx, []byte("=")) == 1 && bytes.Count(tx, []byte(":")) == 0 { + if !bytes.HasPrefix(tx, []byte("=")) && !bytes.HasSuffix(tx, []byte("=")) { + return true + } + } + return false +} + +// PrepareProposal is called when the node is a proposer. Tendermint stages a set of transactions to the application. As the +// KVStore has two accepted formats, `:` and `=`, we modify all instances of `:` with `=` to make it consistent. Note: this is +// quite a trivial example of transaction modification. +// NOTE: we assume that Tendermint will never provide more transactions than can fit in a block. +func (app *Application) PrepareProposal(_ context.Context, req *types.RequestPrepareProposal) (*types.ResponsePrepareProposal, error) { + return &types.ResponsePrepareProposal{Txs: app.substPrepareTx(req.Txs)}, nil +} + +// substPrepareTx substitutes all the transactions prefixed with 'prepare' in the +// proposal for transactions with the prefix stripped. +func (app *Application) substPrepareTx(blockData [][]byte) [][]byte { + txs := make([][]byte, len(blockData)) + for idx, tx := range blockData { + txs[idx] = bytes.Replace(tx, []byte(":"), []byte("="), 1) + } + return txs +} + +// ProcessProposal is called whenever a node receives a complete proposal. It allows the application to validate the proposal. +// For the KVStore we check that each transaction has the valid tx format: +// - Contains one and only one `=` +// - `=` is not the first or last byte. +// - if key is `val` that the validator update transaction is also valid +func (app *Application) ProcessProposal(_ context.Context, req *types.RequestProcessProposal) (*types.ResponseProcessProposal, error) { + for _, tx := range req.Txs { + if len(tx) < 3 || + bytes.Count(tx, []byte("=")) != 1 || + bytes.HasPrefix(tx, []byte("=")) || + bytes.HasSuffix(tx, []byte("=")) { + return &types.ResponseProcessProposal{Status: types.ResponseProcessProposal_REJECT}, nil + } + if isValidatorTx(tx) { + if _, _, err := parseValidatorTx(tx); err != nil { + return &types.ResponseProcessProposal{Status: types.ResponseProcessProposal_REJECT}, nil + } + } + } + return &types.ResponseProcessProposal{Status: types.ResponseProcessProposal_ACCEPT}, nil +} + +// FinalizeBlock executes the block against the application state. It punishes validators who equivocated and +// updates validators according to transactions in a block. The rest of the transactions are regular key value +// updates and are cached in memory and will be persisted once Commit is called. +// ConsensusParams are never changed. func (app *Application) FinalizeBlock(_ context.Context, req *types.RequestFinalizeBlock) (*types.ResponseFinalizeBlock, error) { // reset valset changes - app.ValUpdates = make([]types.ValidatorUpdate, 0) + app.valUpdates = make([]types.ValidatorUpdate, 0) + app.stagedTxs = make([][]byte, 0) // Punish validators who committed equivocation. for _, ev := range req.Misbehavior { @@ -194,29 +217,61 @@ func (app *Application) FinalizeBlock(_ context.Context, req *types.RequestFinal respTxs := make([]*types.ExecTxResult, len(req.Txs)) for i, tx := range req.Txs { - respTxs[i] = app.handleTx(tx) + if isValidatorTx(tx) { + pubKey, power, err := parseValidatorTx(tx) + if err != nil { + panic(err) + } + app.valUpdates = append(app.valUpdates, types.UpdateValidator(pubKey, power, "")) + } else { + app.stagedTxs = append(app.stagedTxs, tx) + } + respTxs[i] = &types.ExecTxResult{ + Code: CodeTypeOK, + // With every transaction we can emit a series of events. To make it simple, we just emit the same events. + Events: []types.Event{ + { + Type: "app", + Attributes: []types.EventAttribute{ + {Key: "creator", Value: "Cosmoshi Netowoko", Index: true}, + {Key: "index_key", Value: "index is working", Index: true}, + {Key: "noindex_key", Value: "index is working", Index: false}, + }, + }, + }, + } + app.state.Size++ } - // Using a memdb - just return the big endian size of the db - appHash := make([]byte, 8) - binary.PutVarint(appHash, app.state.Size) - app.state.AppHash = appHash - app.state.Height++ + app.state.Height = req.Height - return &types.ResponseFinalizeBlock{TxResults: respTxs, ValidatorUpdates: app.ValUpdates, AgreedAppData: appHash}, nil + return &types.ResponseFinalizeBlock{TxResults: respTxs, ValidatorUpdates: app.valUpdates, AgreedAppData: app.state.Hash()}, nil } -func (app *Application) CheckTx(_ context.Context, req *types.RequestCheckTx) (*types.ResponseCheckTx, error) { - if req.Type == types.CheckTxType_Recheck { - if _, ok := app.txToRemove[string(req.Tx)]; ok { - return &types.ResponseCheckTx{Code: CodeTypeExecuted, GasWanted: 1}, nil +// Commit is called after FinalizeBlock and after Tendermint state which includes the updates to +// AppHash, ConsensusParams and ValidatorSet has occurred. +// The KVStore persists the validator updates and the new key values +func (app *Application) Commit(_ context.Context, _ *types.RequestCommit) (*types.ResponseCommit, error) { + + // apply the validator updates to state (note this is really the validator set at h + 2) + for _, valUpdate := range app.valUpdates { + app.updateValidator(valUpdate) + } + + // persist all the staged txs in the kvstore + for _, tx := range app.stagedTxs { + parts := bytes.Split(tx, []byte("=")) + if len(parts) != 2 { + panic(fmt.Sprintf("unexpected tx format. Expected 2 got %d: %s", len(parts), parts)) + } + key, value := string(parts[0]), string(parts[1]) + err := app.state.db.Set(prefixKey([]byte(key)), []byte(value)) + if err != nil { + panic(err) } } - return &types.ResponseCheckTx{Code: CodeTypeOK, GasWanted: 1}, nil -} -func (app *Application) Commit(_ context.Context, _ *types.RequestCommit) (*types.ResponseCommit, error) { - // empty out the set of transactions to remove via rechecktx + // persist the state (i.e. size and height) saveState(app.state) resp := &types.ResponseCommit{} @@ -231,7 +286,7 @@ func (app *Application) Query(_ context.Context, reqQuery *types.RequestQuery) ( resQuery := &types.ResponseQuery{} if reqQuery.Path == "/val" { - key := []byte("val:" + string(reqQuery.Data)) + key := []byte(ValidatorPrefix + string(reqQuery.Data)) value, err := app.state.db.Get(key) if err != nil { panic(err) @@ -278,23 +333,79 @@ func (app *Application) Query(_ context.Context, reqQuery *types.RequestQuery) ( return resQuery, nil } -func (app *Application) PrepareProposal(_ context.Context, req *types.RequestPrepareProposal) (*types.ResponsePrepareProposal, error) { - return &types.ResponsePrepareProposal{Txs: app.substPrepareTx(req.Txs, req.MaxTxBytes)}, nil +func (app *Application) Close() error { + return app.state.db.Close() } -func (app *Application) ProcessProposal(_ context.Context, req *types.RequestProcessProposal) (*types.ResponseProcessProposal, error) { - for _, tx := range req.Txs { - if len(tx) == 0 || isPrepareTx(tx) { - return &types.ResponseProcessProposal{Status: types.ResponseProcessProposal_REJECT}, nil - } +func isValidatorTx(tx []byte) bool { + return strings.HasPrefix(string(tx), ValidatorPrefix) +} + +func parseValidatorTx(tx []byte) ([]byte, int64, error) { + tx = tx[len(ValidatorPrefix):] + + // get the pubkey and power + pubKeyAndPower := strings.Split(string(tx), "!") + if len(pubKeyAndPower) != 2 { + return nil, 0, fmt.Errorf("Expected 'pubkey!power'. Got %v", pubKeyAndPower) } - return &types.ResponseProcessProposal{Status: types.ResponseProcessProposal_ACCEPT}, nil + pubkeyS, powerS := pubKeyAndPower[0], pubKeyAndPower[1] + + // decode the pubkey + pubkey, err := base64.StdEncoding.DecodeString(pubkeyS) + if err != nil { + return nil, 0, fmt.Errorf("Pubkey (%s) is invalid base64", pubkeyS) + } + + // decode the power + power, err := strconv.ParseInt(powerS, 10, 64) + if err != nil { + return nil, 0, fmt.Errorf("Power (%s) is not an int", powerS) + } + + if power < 0 { + return nil, 0, fmt.Errorf("Power can not be less than 0, got %d", power) + } + + return pubkey, power, nil } -//--------------------------------------------- -// update validators +// add, update, or remove a validator +func (app *Application) updateValidator(v types.ValidatorUpdate) { + pubkey, err := cryptoencoding.PubKeyFromProto(v.PubKey) + if err != nil { + panic(fmt.Errorf("can't decode public key: %w", err)) + } + key := []byte(ValidatorPrefix + string(pubkey.Bytes())) -func (app *Application) Validators() (validators []types.ValidatorUpdate) { + if v.Power == 0 { + // remove validator + hasKey, err := app.state.db.Has(key) + if err != nil { + panic(err) + } + if !hasKey { + pubStr := base64.StdEncoding.EncodeToString(pubkey.Bytes()) + panic(fmt.Sprintf("Cannot remove non-existent validator %s", pubStr)) + } + if err = app.state.db.Delete(key); err != nil { + panic(err) + } + delete(app.valAddrToPubKeyMap, string(pubkey.Address())) + } else { + // add or update validator + value := bytes.NewBuffer(make([]byte, 0)) + if err := types.WriteMessage(&v, value); err != nil { + panic(err) + } + if err = app.state.db.Set(key, value.Bytes()); err != nil { + panic(err) + } + app.valAddrToPubKeyMap[string(pubkey.Address())] = v.PubKey + } +} + +func (app *Application) getValidators() (validators []types.ValidatorUpdate) { itr, err := app.state.db.Iterator(nil, nil) if err != nil { panic(err) @@ -315,130 +426,54 @@ func (app *Application) Validators() (validators []types.ValidatorUpdate) { return } -func MakeValSetChangeTx(pubkey cryptoproto.PublicKey, power int64) []byte { - pk, err := cryptoencoding.PubKeyFromProto(pubkey) +// ----------------------------- + +type State struct { + db dbm.DB + // Size is essentially the amount of transactions that have been processes. + // This is used for the appHash + Size int64 `json:"size"` + Height int64 `json:"height"` +} + +func loadState(db dbm.DB) State { + var state State + state.db = db + stateBytes, err := db.Get(stateKey) if err != nil { panic(err) } - pubStr := base64.StdEncoding.EncodeToString(pk.Bytes()) - return []byte(fmt.Sprintf("val:%s!%d", pubStr, power)) -} - -func isValidatorTx(tx []byte) bool { - return strings.HasPrefix(string(tx), ValidatorSetChangePrefix) -} - -// format is "val:pubkey!power" -// pubkey is a base64-encoded 32-byte ed25519 key -func (app *Application) execValidatorTx(tx []byte) *types.ExecTxResult { - tx = tx[len(ValidatorSetChangePrefix):] - - // get the pubkey and power - pubKeyAndPower := strings.Split(string(tx), "!") - if len(pubKeyAndPower) != 2 { - return &types.ExecTxResult{ - Code: CodeTypeEncodingError, - Log: fmt.Sprintf("Expected 'pubkey!power'. Got %v", pubKeyAndPower)} + if len(stateBytes) == 0 { + return state } - pubkeyS, powerS := pubKeyAndPower[0], pubKeyAndPower[1] - - // decode the pubkey - pubkey, err := base64.StdEncoding.DecodeString(pubkeyS) + err = json.Unmarshal(stateBytes, &state) if err != nil { - return &types.ExecTxResult{ - Code: CodeTypeEncodingError, - Log: fmt.Sprintf("Pubkey (%s) is invalid base64", pubkeyS)} + panic(err) } + return state +} - // decode the power - power, err := strconv.ParseInt(powerS, 10, 64) +func saveState(state State) { + stateBytes, err := json.Marshal(state) if err != nil { - return &types.ExecTxResult{ - Code: CodeTypeEncodingError, - Log: fmt.Sprintf("Power (%s) is not an int", powerS)} + panic(err) } - - // update - return app.updateValidator(types.UpdateValidator(pubkey, power, "")) -} - -// add, update, or remove a validator -func (app *Application) updateValidator(v types.ValidatorUpdate) *types.ExecTxResult { - pubkey, err := cryptoencoding.PubKeyFromProto(v.PubKey) + err = state.db.Set(stateKey, stateBytes) if err != nil { - panic(fmt.Errorf("can't decode public key: %w", err)) + panic(err) } - key := []byte("val:" + string(pubkey.Bytes())) - - if v.Power == 0 { - // remove validator - hasKey, err := app.state.db.Has(key) - if err != nil { - panic(err) - } - if !hasKey { - pubStr := base64.StdEncoding.EncodeToString(pubkey.Bytes()) - return &types.ExecTxResult{ - Code: CodeTypeUnauthorized, - Log: fmt.Sprintf("Cannot remove non-existent validator %s", pubStr)} - } - if err = app.state.db.Delete(key); err != nil { - panic(err) - } - delete(app.valAddrToPubKeyMap, string(pubkey.Address())) - } else { - // add or update validator - value := bytes.NewBuffer(make([]byte, 0)) - if err := types.WriteMessage(&v, value); err != nil { - return &types.ExecTxResult{ - Code: CodeTypeEncodingError, - Log: fmt.Sprintf("error encoding validator: %v", err)} - } - if err = app.state.db.Set(key, value.Bytes()); err != nil { - panic(err) - } - app.valAddrToPubKeyMap[string(pubkey.Address())] = v.PubKey - } - - // we only update the changes array if we successfully updated the tree - app.ValUpdates = append(app.ValUpdates, v) - - return &types.ExecTxResult{Code: CodeTypeOK} } -// ----------------------------- - -const ( - PreparePrefix = "prepare" - ReplacePrefix = "replace" - - ValidatorSetChangePrefix string = "val:" -) - -func isPrepareTx(tx []byte) bool { return bytes.HasPrefix(tx, []byte(PreparePrefix)) } - -// execPrepareTx is noop. tx data is considered as placeholder -// and is substitute at the PrepareProposal. -func (app *Application) execPrepareTx(tx []byte) *types.ExecTxResult { - // noop - return &types.ExecTxResult{} +// Hash returns the hash of the application state. This is computed +// as the size or number of transactions processed within the state. Note that this isn't +// a strong gaurantee of state machine replication because states could +// have different kv values but still have the same size. +func (s State) Hash() []byte { + appHash := make([]byte, 8) + binary.PutVarint(appHash, s.Size) + return appHash } -// substPrepareTx substitutes all the transactions prefixed with 'prepare' in the -// proposal for transactions with the prefix stripped. -func (app *Application) substPrepareTx(blockData [][]byte, maxTxBytes int64) [][]byte { - txs := make([][]byte, 0, len(blockData)) - var totalBytes int64 - for _, tx := range blockData { - txMod := tx - if isPrepareTx(tx) { - txMod = bytes.Replace(tx, []byte(PreparePrefix), []byte(ReplacePrefix), 1) - } - totalBytes += int64(len(txMod)) - if totalBytes > maxTxBytes { - break - } - txs = append(txs, txMod) - } - return txs +func prefixKey(key []byte) []byte { + return append(kvPairPrefixKey, key...) } diff --git a/abci/example/kvstore/kvstore_test.go b/abci/example/kvstore/kvstore_test.go index 03bf17a26..8245333b7 100644 --- a/abci/example/kvstore/kvstore_test.go +++ b/abci/example/kvstore/kvstore_test.go @@ -26,24 +26,21 @@ func TestKVStoreKV(t *testing.T) { defer cancel() kvstore := NewInMemoryApplication() - key := testKey - value := key - tx := []byte(key) - testKVStore(ctx, t, kvstore, tx, key, value) - - value = testValue - tx = []byte(key + "=" + value) - testKVStore(ctx, t, kvstore, tx, key, value) + tx := []byte(testKey + ":" + testValue) + testKVStore(ctx, t, kvstore, tx, testKey, testValue) + tx = []byte(testKey + "=" + testValue) + testKVStore(ctx, t, kvstore, tx, testKey, testValue) } func testKVStore(ctx context.Context, t *testing.T, app types.Application, tx []byte, key, value string) { - req := &types.RequestFinalizeBlock{Txs: [][]byte{tx}} - ar, err := app.FinalizeBlock(ctx, req) + checkTxResp, err := app.CheckTx(ctx, &types.RequestCheckTx{Tx: tx}) require.NoError(t, err) - require.Equal(t, 1, len(ar.TxResults)) - require.False(t, ar.TxResults[0].IsErr()) - // repeating tx doesn't raise error - ar, err = app.FinalizeBlock(ctx, req) + require.Equal(t, uint32(0), checkTxResp.Code) + + ppResp, err := app.PrepareProposal(ctx, &types.RequestPrepareProposal{Txs: [][]byte{tx}}) + require.Len(t, ppResp.Txs, 1) + req := &types.RequestFinalizeBlock{Height: 1, Txs: ppResp.Txs} + ar, err := app.FinalizeBlock(ctx, req) require.NoError(t, err) require.Equal(t, 1, len(ar.TxResults)) require.False(t, ar.TxResults[0].IsErr()) @@ -85,13 +82,8 @@ func TestPersistentKVStoreKV(t *testing.T) { kvstore := NewPersistentApplication(t.TempDir()) key := testKey - value := key - tx := []byte(key) - testKVStore(ctx, t, kvstore, tx, key, value) - - value = testValue - tx = []byte(key + "=" + value) - testKVStore(ctx, t, kvstore, tx, key, value) + value := testValue + testKVStore(ctx, t, kvstore, NewTx(key, value), key, value) } func TestPersistentKVStoreInfo(t *testing.T) { @@ -141,7 +133,7 @@ func TestValUpdates(t *testing.T) { }) require.NoError(t, err) - vals1, vals2 := vals[:nInit], kvstore.Validators() + vals1, vals2 := vals[:nInit], kvstore.getValidators() valsEqual(t, vals1, vals2) var v1, v2, v3 types.ValidatorUpdate @@ -154,7 +146,7 @@ func TestValUpdates(t *testing.T) { makeApplyBlock(ctx, t, kvstore, 1, diff, tx1, tx2) - vals1, vals2 = vals[:nInit+2], kvstore.Validators() + vals1, vals2 = vals[:nInit+2], kvstore.getValidators() valsEqual(t, vals1, vals2) // remove some validators @@ -170,7 +162,7 @@ func TestValUpdates(t *testing.T) { makeApplyBlock(ctx, t, kvstore, 2, diff, tx1, tx2, tx3) vals1 = append(vals[:nInit-2], vals[nInit+1]) //nolint: gocritic - vals2 = kvstore.Validators() + vals2 = kvstore.getValidators() valsEqual(t, vals1, vals2) // update some validators @@ -186,11 +178,35 @@ func TestValUpdates(t *testing.T) { makeApplyBlock(ctx, t, kvstore, 3, diff, tx1) vals1 = append([]types.ValidatorUpdate{v1}, vals1[1:]...) - vals2 = kvstore.Validators() + vals2 = kvstore.getValidators() valsEqual(t, vals1, vals2) } +func TestCheckTx(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + kvstore := NewInMemoryApplication() + + testCases := []struct { + expCode uint32 + tx []byte + }{ + {CodeTypeOK, NewTx("hello", "world")}, + {CodeTypeInvalidTxFormat, []byte("hello")}, + {CodeTypeOK, []byte("space:jam")}, + {CodeTypeInvalidTxFormat, []byte("=hello")}, + {CodeTypeInvalidTxFormat, []byte("hello=")}, + {CodeTypeOK, []byte("a=b")}, + } + + for idx, tc := range testCases { + resp, err := kvstore.CheckTx(ctx, &types.RequestCheckTx{Tx: tc.tx}) + require.NoError(t, err, idx) + require.Equal(t, tc.expCode, resp.Code, idx) + } +} + func TestClientServer(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -198,14 +214,12 @@ func TestClientServer(t *testing.T) { kvstore := NewInMemoryApplication() client, _, err := makeSocketClientServer(t, kvstore, "kvstore-socket") require.NoError(t, err) - runClientTests(ctx, t, client) // set up grpc app kvstore = NewInMemoryApplication() - gclient, _, err := makeGRPCClientServer(kvstore, "/tmp/kvstore-grpc") + gclient, _, err := makeGRPCClientServer(t, kvstore, t.TempDir()) require.NoError(t, err) - runClientTests(ctx, t, gclient) } @@ -283,7 +297,7 @@ func makeSocketClientServer(t *testing.T, app types.Application, name string) (a return client, server, nil } -func makeGRPCClientServer(app types.Application, name string) (abcicli.Client, service.Service, error) { +func makeGRPCClientServer(t *testing.T, app types.Application, name string) (abcicli.Client, service.Service, error) { // Start the listener socket := fmt.Sprintf("unix://%s.sock", name) logger := log.TestingLogger() @@ -294,67 +308,31 @@ func makeGRPCClientServer(app types.Application, name string) (abcicli.Client, s return nil, nil, err } + t.Cleanup(func() { + if err := server.Stop(); err != nil { + t.Error(err) + } + }) + client := abcicli.NewGRPCClient(socket, true) client.SetLogger(logger.With("module", "abci-client")) if err := client.Start(); err != nil { - if err := server.Stop(); err != nil { - return nil, nil, err - } return nil, nil, err } + + t.Cleanup(func() { + if err := client.Stop(); err != nil { + t.Error(err) + } + }) + return client, server, nil } func runClientTests(ctx context.Context, t *testing.T, client abcicli.Client) { // run some tests.... - key := testKey - value := key - tx := []byte(key) - testClient(ctx, t, client, tx, key, value) - - value = testValue - tx = []byte(key + "=" + value) - testClient(ctx, t, client, tx, key, value) -} - -func testClient(ctx context.Context, t *testing.T, app abcicli.Client, tx []byte, key, value string) { - ar, err := app.FinalizeBlock(ctx, &types.RequestFinalizeBlock{Txs: [][]byte{tx}}) - require.NoError(t, err) - require.Equal(t, 1, len(ar.TxResults)) - require.False(t, ar.TxResults[0].IsErr()) - // repeating FinalizeBlock doesn't raise error - ar, err = app.FinalizeBlock(ctx, &types.RequestFinalizeBlock{Txs: [][]byte{tx}}) - require.NoError(t, err) - require.Equal(t, 1, len(ar.TxResults)) - require.False(t, ar.TxResults[0].IsErr()) - // commit - _, err = app.Commit(ctx, &types.RequestCommit{}) - require.NoError(t, err) - - info, err := app.Info(ctx, &types.RequestInfo{}) - require.NoError(t, err) - require.NotZero(t, info.LastBlockHeight) - - // make sure query is fine - resQuery, err := app.Query(ctx, &types.RequestQuery{ - Path: "/store", - Data: []byte(key), - }) - require.NoError(t, err) - require.Equal(t, CodeTypeOK, resQuery.Code) - require.Equal(t, key, string(resQuery.Key)) - require.Equal(t, value, string(resQuery.Value)) - require.EqualValues(t, info.LastBlockHeight, resQuery.Height) - - // make sure proof is fine - resQuery, err = app.Query(ctx, &types.RequestQuery{ - Path: "/store", - Data: []byte(key), - Prove: true, - }) - require.NoError(t, err) - require.Equal(t, CodeTypeOK, resQuery.Code) - require.Equal(t, key, string(resQuery.Key)) - require.Equal(t, value, string(resQuery.Value)) - require.EqualValues(t, info.LastBlockHeight, resQuery.Height) + tx := []byte(testKey + ":" + testValue) + testKVStore(ctx, t, client, tx, testKey, testValue) + tx = []byte(testKey + "=" + testValue) + testKVStore(ctx, t, client, tx, testKey, testValue) } diff --git a/consensus/byzantine_test.go b/consensus/byzantine_test.go index f90ba13fc..f134af5ac 100644 --- a/consensus/byzantine_test.go +++ b/consensus/byzantine_test.go @@ -478,7 +478,7 @@ func byzantineDecideProposalFunc(t *testing.T, height int64, round int32, cs *St proposal1.Signature = p1.Signature // some new transactions come in (this ensures that the proposals are different) - deliverTxsRange(cs, 0, 1) + deliverTxsRange(t, cs, 0, 1) // Create a new proposal block from state/txs from the mempool. block2, err := cs.createProposalBlock() diff --git a/consensus/common_test.go b/consensus/common_test.go index ab2cc98cb..1d81060b9 100644 --- a/consensus/common_test.go +++ b/consensus/common_test.go @@ -435,6 +435,7 @@ func newStateWithConfigAndBlockStore( stateStore := sm.NewStore(stateDB, sm.StoreOptions{ DiscardFinalizeBlockResponses: false, }) + if err := stateStore.Save(state); err != nil { // for save height 1's validators info panic(err) } diff --git a/consensus/mempool_test.go b/consensus/mempool_test.go index c78d0fc6b..ad8476e06 100644 --- a/consensus/mempool_test.go +++ b/consensus/mempool_test.go @@ -16,6 +16,7 @@ import ( "github.com/tendermint/tendermint/abci/example/kvstore" abci "github.com/tendermint/tendermint/abci/types" mempl "github.com/tendermint/tendermint/mempool" + "github.com/tendermint/tendermint/proxy" sm "github.com/tendermint/tendermint/state" "github.com/tendermint/tendermint/types" ) @@ -30,7 +31,11 @@ func TestMempoolNoProgressUntilTxsAvailable(t *testing.T) { defer os.RemoveAll(config.RootDir) config.Consensus.CreateEmptyBlocks = false state, privVals := randGenesisState(1, false, 10) - cs := newStateWithConfig(config, state, privVals[0], kvstore.NewInMemoryApplication()) + app := kvstore.NewInMemoryApplication() + resp, err := app.Info(context.Background(), proxy.RequestInfo) + require.NoError(t, err) + state.AppHash = resp.LastBlockAppHash + cs := newStateWithConfig(config, state, privVals[0], app) assertMempool(cs.txNotifier).EnableTxsAvailable() height, round := cs.Height, cs.Round newBlockCh := subscribe(cs.eventBus, types.EventQueryNewBlock) @@ -38,7 +43,7 @@ func TestMempoolNoProgressUntilTxsAvailable(t *testing.T) { ensureNewEventOnChannel(newBlockCh) // first block gets committed ensureNoNewEventOnChannel(newBlockCh) - deliverTxsRange(cs, 0, 1) + deliverTxsRange(t, cs, 0, 1) ensureNewEventOnChannel(newBlockCh) // commit txs ensureNewEventOnChannel(newBlockCh) // commit updated app hash ensureNoNewEventOnChannel(newBlockCh) @@ -50,7 +55,11 @@ func TestMempoolProgressAfterCreateEmptyBlocksInterval(t *testing.T) { config.Consensus.CreateEmptyBlocksInterval = ensureTimeout state, privVals := randGenesisState(1, false, 10) - cs := newStateWithConfig(config, state, privVals[0], kvstore.NewInMemoryApplication()) + app := kvstore.NewInMemoryApplication() + resp, err := app.Info(context.Background(), proxy.RequestInfo) + require.NoError(t, err) + state.AppHash = resp.LastBlockAppHash + cs := newStateWithConfig(config, state, privVals[0], app) assertMempool(cs.txNotifier).EnableTxsAvailable() @@ -91,7 +100,7 @@ func TestMempoolProgressInHigherRound(t *testing.T) { round = 0 ensureNewRound(newRoundCh, height, round) // first round at next height - deliverTxsRange(cs, 0, 1) // we deliver txs, but dont set a proposal so we get the next round + deliverTxsRange(t, cs, 0, 1) // we deliver txs, but dont set a proposal so we get the next round ensureNewTimeout(timeoutCh, height, round, cs.config.TimeoutPropose.Nanoseconds()) round++ // moving to the next round @@ -99,15 +108,11 @@ func TestMempoolProgressInHigherRound(t *testing.T) { ensureNewEventOnChannel(newBlockCh) // now we can commit the block } -func deliverTxsRange(cs *State, start, end int) { +func deliverTxsRange(t *testing.T, cs *State, start, end int) { // Deliver some txs. for i := start; i < end; i++ { - txBytes := make([]byte, 8) - binary.BigEndian.PutUint64(txBytes, uint64(i)) - err := assertMempool(cs.txNotifier).CheckTx(txBytes, nil, mempl.TxInfo{}) - if err != nil { - panic(fmt.Sprintf("Error after CheckTx: %v", err)) - } + err := assertMempool(cs.txNotifier).CheckTx(kvstore.NewTx(fmt.Sprintf("%d", i), "true"), nil, mempl.TxInfo{}) + require.NoError(t, err) } } @@ -121,7 +126,7 @@ func TestMempoolTxConcurrentWithCommit(t *testing.T) { newBlockEventsCh := subscribe(cs.eventBus, types.EventQueryNewBlockEvents) const numTxs int64 = 3000 - go deliverTxsRange(cs, 0, int(numTxs)) + go deliverTxsRange(t, cs, 0, int(numTxs)) startTestRound(cs, cs.Height, cs.Round) for n := int64(0); n < numTxs; { @@ -148,7 +153,7 @@ func TestMempoolRmBadTx(t *testing.T) { txBytes := make([]byte, 8) binary.BigEndian.PutUint64(txBytes, uint64(0)) - res, err := app.FinalizeBlock(context.Background(), &abci.RequestFinalizeBlock{Txs: [][]byte{txBytes}}) + res, err := app.FinalizeBlock(context.Background(), &abci.RequestFinalizeBlock{Txs: [][]byte{kvstore.NewTx("key", "value")}}) require.NoError(t, err) assert.False(t, res.TxResults[0].IsErr()) assert.True(t, len(res.AgreedAppData) > 0) @@ -160,7 +165,7 @@ func TestMempoolRmBadTx(t *testing.T) { // CheckTx should not err, but the app should return a bad abci code // and the tx should get removed from the pool err := assertMempool(cs.txNotifier).CheckTx(txBytes, func(r *abci.ResponseCheckTx) { - if r.Code != kvstore.CodeTypeBadNonce { + if r.Code != kvstore.CodeTypeInvalidTxFormat { t.Errorf("expected checktx to return bad nonce, got %v", r) return } diff --git a/types/events.go b/types/events.go index 960465cc2..db6860504 100644 --- a/types/events.go +++ b/types/events.go @@ -177,7 +177,7 @@ func QueryForEvent(eventType string) tmpubsub.Query { type BlockEventPublisher interface { PublishEventNewBlock(block EventDataNewBlock) error PublishEventNewBlockHeader(header EventDataNewBlockHeader) error - PublishEventNewBlockEvents(header EventDataNewBlockEvents) error + PublishEventNewBlockEvents(events EventDataNewBlockEvents) error PublishEventNewEvidence(evidence EventDataNewEvidence) error PublishEventTx(EventDataTx) error PublishEventValidatorSetUpdates(EventDataValidatorSetUpdates) error