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.
362 lines
10 KiB
362 lines
10 KiB
package app
|
|
|
|
import (
|
|
"bytes"
|
|
"encoding/base64"
|
|
"errors"
|
|
"fmt"
|
|
"path/filepath"
|
|
"sort"
|
|
"strconv"
|
|
"sync"
|
|
|
|
"github.com/tendermint/tendermint/abci/example/code"
|
|
abci "github.com/tendermint/tendermint/abci/types"
|
|
"github.com/tendermint/tendermint/libs/log"
|
|
"github.com/tendermint/tendermint/proto/tendermint/types"
|
|
"github.com/tendermint/tendermint/version"
|
|
)
|
|
|
|
// Application is an ABCI application for use by end-to-end tests. It is a
|
|
// simple key/value store for strings, storing data in memory and persisting
|
|
// to disk as JSON, taking state sync snapshots if requested.
|
|
type Application struct {
|
|
abci.BaseApplication
|
|
mu sync.Mutex
|
|
logger log.Logger
|
|
state *State
|
|
snapshots *SnapshotStore
|
|
cfg *Config
|
|
restoreSnapshot *abci.Snapshot
|
|
restoreChunks [][]byte
|
|
}
|
|
|
|
// Config allows for the setting of high level parameters for running the e2e Application
|
|
// KeyType and ValidatorUpdates must be the same for all nodes running the same application.
|
|
type Config struct {
|
|
// The directory with which state.json will be persisted in. Usually $HOME/.tendermint/data
|
|
Dir string `toml:"dir"`
|
|
|
|
// SnapshotInterval specifies the height interval at which the application
|
|
// will take state sync snapshots. Defaults to 0 (disabled).
|
|
SnapshotInterval uint64 `toml:"snapshot_interval"`
|
|
|
|
// RetainBlocks specifies the number of recent blocks to retain. Defaults to
|
|
// 0, which retains all blocks. Must be greater that PersistInterval,
|
|
// SnapshotInterval and EvidenceAgeHeight.
|
|
RetainBlocks uint64 `toml:"retain_blocks"`
|
|
|
|
// KeyType sets the curve that will be used by validators.
|
|
// Options are ed25519 & secp256k1
|
|
KeyType string `toml:"key_type"`
|
|
|
|
// PersistInterval specifies the height interval at which the application
|
|
// will persist state to disk. Defaults to 1 (every height), setting this to
|
|
// 0 disables state persistence.
|
|
PersistInterval uint64 `toml:"persist_interval"`
|
|
|
|
// ValidatorUpdates is a map of heights to validator names and their power,
|
|
// and will be returned by the ABCI application. For example, the following
|
|
// changes the power of validator01 and validator02 at height 1000:
|
|
//
|
|
// [validator_update.1000]
|
|
// validator01 = 20
|
|
// validator02 = 10
|
|
//
|
|
// Specifying height 0 returns the validator update during InitChain. The
|
|
// application returns the validator updates as-is, i.e. removing a
|
|
// validator must be done by returning it with power 0, and any validators
|
|
// not specified are not changed.
|
|
//
|
|
// height <-> pubkey <-> voting power
|
|
ValidatorUpdates map[string]map[string]uint8 `toml:"validator_update"`
|
|
}
|
|
|
|
func DefaultConfig(dir string) *Config {
|
|
return &Config{
|
|
PersistInterval: 1,
|
|
SnapshotInterval: 100,
|
|
Dir: dir,
|
|
}
|
|
}
|
|
|
|
// NewApplication creates the application.
|
|
func NewApplication(cfg *Config) (*Application, error) {
|
|
state, err := NewState(cfg.Dir, cfg.PersistInterval)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
snapshots, err := NewSnapshotStore(filepath.Join(cfg.Dir, "snapshots"))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &Application{
|
|
logger: log.MustNewDefaultLogger(log.LogFormatPlain, log.LogLevelInfo),
|
|
state: state,
|
|
snapshots: snapshots,
|
|
cfg: cfg,
|
|
}, nil
|
|
}
|
|
|
|
// Info implements ABCI.
|
|
func (app *Application) Info(req abci.RequestInfo) abci.ResponseInfo {
|
|
app.mu.Lock()
|
|
defer app.mu.Unlock()
|
|
|
|
return abci.ResponseInfo{
|
|
Version: version.ABCIVersion,
|
|
AppVersion: 1,
|
|
LastBlockHeight: int64(app.state.Height),
|
|
LastBlockAppHash: app.state.Hash,
|
|
}
|
|
}
|
|
|
|
// Info implements ABCI.
|
|
func (app *Application) InitChain(req abci.RequestInitChain) abci.ResponseInitChain {
|
|
app.mu.Lock()
|
|
defer app.mu.Unlock()
|
|
|
|
var err error
|
|
app.state.initialHeight = uint64(req.InitialHeight)
|
|
if len(req.AppStateBytes) > 0 {
|
|
err = app.state.Import(0, req.AppStateBytes)
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
}
|
|
resp := abci.ResponseInitChain{
|
|
AppHash: app.state.Hash,
|
|
ConsensusParams: &types.ConsensusParams{
|
|
Version: &types.VersionParams{
|
|
AppVersion: 1,
|
|
},
|
|
},
|
|
}
|
|
if resp.Validators, err = app.validatorUpdates(0); err != nil {
|
|
panic(err)
|
|
}
|
|
return resp
|
|
}
|
|
|
|
// CheckTx implements ABCI.
|
|
func (app *Application) CheckTx(req abci.RequestCheckTx) abci.ResponseCheckTx {
|
|
app.mu.Lock()
|
|
defer app.mu.Unlock()
|
|
|
|
_, _, err := parseTx(req.Tx)
|
|
if err != nil {
|
|
return abci.ResponseCheckTx{
|
|
Code: code.CodeTypeEncodingError,
|
|
Log: err.Error(),
|
|
}
|
|
}
|
|
return abci.ResponseCheckTx{Code: code.CodeTypeOK, GasWanted: 1}
|
|
}
|
|
|
|
// FinalizeBlock implements ABCI.
|
|
func (app *Application) FinalizeBlock(req abci.RequestFinalizeBlock) abci.ResponseFinalizeBlock {
|
|
var txs = make([]*abci.ResponseDeliverTx, len(req.Txs))
|
|
|
|
app.mu.Lock()
|
|
defer app.mu.Unlock()
|
|
|
|
for i, tx := range req.Txs {
|
|
key, value, err := parseTx(tx)
|
|
if err != nil {
|
|
panic(err) // shouldn't happen since we verified it in CheckTx
|
|
}
|
|
app.state.Set(key, value)
|
|
|
|
txs[i] = &abci.ResponseDeliverTx{Code: code.CodeTypeOK}
|
|
}
|
|
|
|
valUpdates, err := app.validatorUpdates(uint64(req.Height))
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
|
|
return abci.ResponseFinalizeBlock{
|
|
Txs: txs,
|
|
ValidatorUpdates: valUpdates,
|
|
Events: []abci.Event{
|
|
{
|
|
Type: "val_updates",
|
|
Attributes: []abci.EventAttribute{
|
|
{
|
|
Key: "size",
|
|
Value: strconv.Itoa(valUpdates.Len()),
|
|
},
|
|
{
|
|
Key: "height",
|
|
Value: strconv.Itoa(int(req.Height)),
|
|
},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
}
|
|
|
|
// Commit implements ABCI.
|
|
func (app *Application) Commit() abci.ResponseCommit {
|
|
app.mu.Lock()
|
|
defer app.mu.Unlock()
|
|
|
|
height, hash, err := app.state.Commit()
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
if app.cfg.SnapshotInterval > 0 && height%app.cfg.SnapshotInterval == 0 {
|
|
snapshot, err := app.snapshots.Create(app.state)
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
app.logger.Info("Created state sync snapshot", "height", snapshot.Height)
|
|
err = app.snapshots.Prune(maxSnapshotCount)
|
|
if err != nil {
|
|
app.logger.Error("Failed to prune snapshots", "err", err)
|
|
}
|
|
}
|
|
retainHeight := int64(0)
|
|
if app.cfg.RetainBlocks > 0 {
|
|
retainHeight = int64(height - app.cfg.RetainBlocks + 1)
|
|
}
|
|
return abci.ResponseCommit{
|
|
Data: hash,
|
|
RetainHeight: retainHeight,
|
|
}
|
|
}
|
|
|
|
// Query implements ABCI.
|
|
func (app *Application) Query(req abci.RequestQuery) abci.ResponseQuery {
|
|
app.mu.Lock()
|
|
defer app.mu.Unlock()
|
|
|
|
return abci.ResponseQuery{
|
|
Height: int64(app.state.Height),
|
|
Key: req.Data,
|
|
Value: []byte(app.state.Get(string(req.Data))),
|
|
}
|
|
}
|
|
|
|
// ListSnapshots implements ABCI.
|
|
func (app *Application) ListSnapshots(req abci.RequestListSnapshots) abci.ResponseListSnapshots {
|
|
app.mu.Lock()
|
|
defer app.mu.Unlock()
|
|
|
|
snapshots, err := app.snapshots.List()
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
return abci.ResponseListSnapshots{Snapshots: snapshots}
|
|
}
|
|
|
|
// LoadSnapshotChunk implements ABCI.
|
|
func (app *Application) LoadSnapshotChunk(req abci.RequestLoadSnapshotChunk) abci.ResponseLoadSnapshotChunk {
|
|
app.mu.Lock()
|
|
defer app.mu.Unlock()
|
|
|
|
chunk, err := app.snapshots.LoadChunk(req.Height, req.Format, req.Chunk)
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
return abci.ResponseLoadSnapshotChunk{Chunk: chunk}
|
|
}
|
|
|
|
// OfferSnapshot implements ABCI.
|
|
func (app *Application) OfferSnapshot(req abci.RequestOfferSnapshot) abci.ResponseOfferSnapshot {
|
|
app.mu.Lock()
|
|
defer app.mu.Unlock()
|
|
|
|
if app.restoreSnapshot != nil {
|
|
panic("A snapshot is already being restored")
|
|
}
|
|
app.restoreSnapshot = req.Snapshot
|
|
app.restoreChunks = [][]byte{}
|
|
return abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_ACCEPT}
|
|
}
|
|
|
|
// ApplySnapshotChunk implements ABCI.
|
|
func (app *Application) ApplySnapshotChunk(req abci.RequestApplySnapshotChunk) abci.ResponseApplySnapshotChunk {
|
|
app.mu.Lock()
|
|
defer app.mu.Unlock()
|
|
|
|
if app.restoreSnapshot == nil {
|
|
panic("No restore in progress")
|
|
}
|
|
app.restoreChunks = append(app.restoreChunks, req.Chunk)
|
|
if len(app.restoreChunks) == int(app.restoreSnapshot.Chunks) {
|
|
bz := []byte{}
|
|
for _, chunk := range app.restoreChunks {
|
|
bz = append(bz, chunk...)
|
|
}
|
|
err := app.state.Import(app.restoreSnapshot.Height, bz)
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
app.restoreSnapshot = nil
|
|
app.restoreChunks = nil
|
|
}
|
|
return abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}
|
|
}
|
|
|
|
func (app *Application) PrepareProposal(req abci.RequestPrepareProposal) abci.ResponsePrepareProposal {
|
|
return abci.ResponsePrepareProposal{BlockData: req.BlockData}
|
|
}
|
|
|
|
// ProcessProposal implements part of the Application interface.
|
|
// It accepts any proposal that does not contain a malformed transaction.
|
|
func (app *Application) ProcessProposal(req abci.RequestProcessProposal) abci.ResponseProcessProposal {
|
|
for _, tx := range req.Txs {
|
|
_, _, err := parseTx(tx)
|
|
if err != nil {
|
|
return abci.ResponseProcessProposal{Accept: false}
|
|
}
|
|
}
|
|
return abci.ResponseProcessProposal{Accept: true}
|
|
}
|
|
|
|
func (app *Application) Rollback() error {
|
|
app.mu.Lock()
|
|
defer app.mu.Unlock()
|
|
|
|
return app.state.Rollback()
|
|
}
|
|
|
|
// validatorUpdates generates a validator set update.
|
|
func (app *Application) validatorUpdates(height uint64) (abci.ValidatorUpdates, error) {
|
|
updates := app.cfg.ValidatorUpdates[fmt.Sprintf("%v", height)]
|
|
if len(updates) == 0 {
|
|
return nil, nil
|
|
}
|
|
|
|
valUpdates := abci.ValidatorUpdates{}
|
|
for keyString, power := range updates {
|
|
|
|
keyBytes, err := base64.StdEncoding.DecodeString(keyString)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("invalid base64 pubkey value %q: %w", keyString, err)
|
|
}
|
|
valUpdates = append(valUpdates, abci.UpdateValidator(keyBytes, int64(power), app.cfg.KeyType))
|
|
}
|
|
|
|
// the validator updates could be returned in arbitrary order,
|
|
// and that seems potentially bad. This orders the validator
|
|
// set.
|
|
sort.Slice(valUpdates, func(i, j int) bool {
|
|
return valUpdates[i].PubKey.Compare(valUpdates[j].PubKey) < 0
|
|
})
|
|
|
|
return valUpdates, nil
|
|
}
|
|
|
|
// parseTx parses a tx in 'key=value' format into a key and value.
|
|
func parseTx(tx []byte) (string, string, error) {
|
|
parts := bytes.Split(tx, []byte("="))
|
|
if len(parts) != 2 {
|
|
return "", "", fmt.Errorf("invalid tx format: %q", string(tx))
|
|
}
|
|
if len(parts[0]) == 0 {
|
|
return "", "", errors.New("key cannot be empty")
|
|
}
|
|
return string(parts[0]), string(parts[1]), nil
|
|
}
|