Browse Source

state: persist validators

pull/618/head
Ethan Buchman 7 years ago
parent
commit
78446fd99c
5 changed files with 161 additions and 28 deletions
  1. +3
    -0
      consensus/replay.go
  2. +8
    -0
      state/errors.go
  3. +3
    -0
      state/execution.go
  4. +101
    -14
      state/state.go
  5. +46
    -14
      state/state_test.go

+ 3
- 0
consensus/replay.go View File

@ -324,8 +324,11 @@ func (h *Handshaker) ReplayBlocks(appHash []byte, appBlockHeight int, proxyApp p
func (h *Handshaker) replayBlocks(proxyApp proxy.AppConns, appBlockHeight, storeBlockHeight int, mutateState bool) ([]byte, error) { func (h *Handshaker) replayBlocks(proxyApp proxy.AppConns, appBlockHeight, storeBlockHeight int, mutateState bool) ([]byte, error) {
// App is further behind than it should be, so we need to replay blocks. // App is further behind than it should be, so we need to replay blocks.
// We replay all blocks from appBlockHeight+1. // We replay all blocks from appBlockHeight+1.
//
// Note that we don't have an old version of the state, // Note that we don't have an old version of the state,
// so we by-pass state validation/mutation using sm.ExecCommitBlock. // so we by-pass state validation/mutation using sm.ExecCommitBlock.
// This also means we won't be saving validator sets if they change during this period.
//
// If mutateState == true, the final block is replayed with h.replayBlock() // If mutateState == true, the final block is replayed with h.replayBlock()
var appHash []byte var appHash []byte


+ 8
- 0
state/errors.go View File

@ -33,6 +33,10 @@ type (
Got *State Got *State
Expected *State Expected *State
} }
ErrNoValSetForHeight struct {
Height int
}
) )
func (e ErrUnknownBlock) Error() string { func (e ErrUnknownBlock) Error() string {
@ -53,3 +57,7 @@ func (e ErrLastStateMismatch) Error() string {
func (e ErrStateMismatch) Error() string { func (e ErrStateMismatch) Error() string {
return cmn.Fmt("State after replay does not match saved state. Got ----\n%v\nExpected ----\n%v\n", e.Got, e.Expected) return cmn.Fmt("State after replay does not match saved state. Got ----\n%v\nExpected ----\n%v\n", e.Got, e.Expected)
} }
func (e ErrNoValSetForHeight) Error() string {
return cmn.Fmt("Could not find validator set for height #%d", e.Height)
}

+ 3
- 0
state/execution.go View File

@ -231,6 +231,9 @@ func (s *State) ApplyBlock(eventCache types.Fireable, proxyAppConn proxy.AppConn
// now update the block and validators // now update the block and validators
s.SetBlockAndValidators(block.Header, partsHeader, abciResponses) s.SetBlockAndValidators(block.Header, partsHeader, abciResponses)
// save the validators for the next block now that we know them
s.SaveValidators()
// lock mempool, commit state, update mempoool // lock mempool, commit state, update mempoool
err = s.CommitStateUpdateMempool(proxyAppConn, block, mempool) err = s.CommitStateUpdateMempool(proxyAppConn, block, mempool)
if err != nil { if err != nil {


+ 101
- 14
state/state.go View File

@ -22,6 +22,10 @@ var (
abciResponsesKey = []byte("abciResponsesKey") abciResponsesKey = []byte("abciResponsesKey")
) )
func calcValidatorsKey(height int) []byte {
return []byte(cmn.Fmt("validatorsKey:%v", height))
}
//----------------------------------------------------------------------------- //-----------------------------------------------------------------------------
// State represents the latest committed state of the Tendermint consensus, // State represents the latest committed state of the Tendermint consensus,
@ -47,8 +51,11 @@ type State struct {
// AppHash is updated after Commit // AppHash is updated after Commit
AppHash []byte AppHash []byte
// XXX: do we need this json tag ?
TxIndexer txindex.TxIndexer `json:"-"` // Transaction indexer. TxIndexer txindex.TxIndexer `json:"-"` // Transaction indexer.
lastHeightValidatorsChanged int
logger log.Logger logger log.Logger
} }
@ -71,6 +78,14 @@ func loadState(db dbm.DB, key []byte) *State {
} }
// TODO: ensure that buf is completely read. // TODO: ensure that buf is completely read.
} }
v := s.loadValidators(s.LastBlockHeight)
if v != nil {
s.lastHeightValidatorsChanged = v.LastHeightChanged
} else {
s.lastHeightValidatorsChanged = 1
}
return s return s
} }
@ -93,6 +108,8 @@ func (s *State) Copy() *State {
AppHash: s.AppHash, AppHash: s.AppHash,
TxIndexer: s.TxIndexer, // pointer here, not value TxIndexer: s.TxIndexer, // pointer here, not value
logger: s.logger, logger: s.logger,
lastHeightValidatorsChanged: s.lastHeightValidatorsChanged,
} }
} }
@ -126,6 +143,56 @@ func (s *State) LoadABCIResponses() *ABCIResponses {
return abciResponses return abciResponses
} }
// SaveValidators persists the validator set for the next block to disk.
// It should be called after the validator set is updated with the results of EndBlock.
// If the validator set did not change after processing the latest block,
// only the last height for which the validators changed is persisted.
func (s *State) SaveValidators() {
lastHeight := s.lastHeightValidatorsChanged
nextHeight := s.LastBlockHeight + 1
v := &Validators{
LastHeightChanged: lastHeight,
}
if lastHeight == nextHeight {
v.ValidatorSet = s.Validators
}
s.db.SetSync(calcValidatorsKey(nextHeight), v.Bytes())
}
// LoadValidators loads the ValidatorSet for a given height.
func (s *State) LoadValidators(height int) (*types.ValidatorSet, error) {
v := s.loadValidators(height)
if v == nil {
return nil, ErrNoValSetForHeight{height}
}
if v.ValidatorSet == nil {
v = s.loadValidators(v.LastHeightChanged)
if v == nil {
return nil, ErrNoValSetForHeight{height}
}
}
return v.ValidatorSet, nil
}
func (s *State) loadValidators(height int) *Validators {
buf := s.db.Get(calcValidatorsKey(height))
if len(buf) == 0 {
return nil
}
v := new(Validators)
r, n, err := bytes.NewReader(buf), new(int), new(error)
wire.ReadBinaryPtr(v, r, 0, n, err)
if *err != nil {
// DATA HAS BEEN CORRUPTED OR THE SPEC HAS CHANGED
cmn.Exit(cmn.Fmt("LoadValidators: Data has been corrupted or its spec has changed: %v\n", *err))
}
// TODO: ensure that buf is completely read.
return v
}
// Equals returns true if the States are identical. // Equals returns true if the States are identical.
func (s *State) Equals(s2 *State) bool { func (s *State) Equals(s2 *State) bool {
return bytes.Equal(s.Bytes(), s2.Bytes()) return bytes.Equal(s.Bytes(), s2.Bytes())
@ -144,10 +211,15 @@ func (s *State) SetBlockAndValidators(header *types.Header, blockPartsHeader typ
prevValSet := s.Validators.Copy() prevValSet := s.Validators.Copy()
nextValSet := prevValSet.Copy() nextValSet := prevValSet.Copy()
err := updateValidators(nextValSet, abciResponses.EndBlock.Diffs)
if err != nil {
s.logger.Error("Error changing validator set", "err", err)
// TODO: err or carry on?
// update the validator set with the latest abciResponses
if len(abciResponses.EndBlock.Diffs) > 0 {
err := updateValidators(nextValSet, abciResponses.EndBlock.Diffs)
if err != nil {
s.logger.Error("Error changing validator set", "err", err)
// TODO: err or carry on?
}
// change results from this height but only applies to the next height
s.lastHeightValidatorsChanged = header.Height + 1
} }
// Update validator accums and set state variables // Update validator accums and set state variables
@ -182,6 +254,7 @@ func GetState(stateDB dbm.DB, genesisFile string) *State {
state := LoadState(stateDB) state := LoadState(stateDB)
if state == nil { if state == nil {
state = MakeGenesisStateFromFile(stateDB, genesisFile) state = MakeGenesisStateFromFile(stateDB, genesisFile)
state.SaveValidators() // save the validators right away for height 1
state.Save() state.Save()
} }
@ -215,6 +288,19 @@ func (a *ABCIResponses) Bytes() []byte {
return wire.BinaryBytes(*a) return wire.BinaryBytes(*a)
} }
//-----------------------------------------------------------------------------
// Validators represents the latest validator set, or the last time it changed
type Validators struct {
ValidatorSet *types.ValidatorSet
LastHeightChanged int
}
// Bytes serializes the Validators using go-wire
func (v *Validators) Bytes() []byte {
return wire.BinaryBytes(*v)
}
//----------------------------------------------------------------------------- //-----------------------------------------------------------------------------
// Genesis // Genesis
@ -260,15 +346,16 @@ func MakeGenesisState(db dbm.DB, genDoc *types.GenesisDoc) *State {
} }
return &State{ return &State{
db: db,
GenesisDoc: genDoc,
ChainID: genDoc.ChainID,
LastBlockHeight: 0,
LastBlockID: types.BlockID{},
LastBlockTime: genDoc.GenesisTime,
Validators: types.NewValidatorSet(validators),
LastValidators: types.NewValidatorSet(nil),
AppHash: genDoc.AppHash,
TxIndexer: &null.TxIndex{}, // we do not need indexer during replay and in tests
db: db,
GenesisDoc: genDoc,
ChainID: genDoc.ChainID,
LastBlockHeight: 0,
LastBlockID: types.BlockID{},
LastBlockTime: genDoc.GenesisTime,
Validators: types.NewValidatorSet(validators),
LastValidators: types.NewValidatorSet(nil),
AppHash: genDoc.AppHash,
TxIndexer: &null.TxIndex{}, // we do not need indexer during replay and in tests
lastHeightValidatorsChanged: 1,
} }
} }

+ 46
- 14
state/state_test.go View File

@ -1,18 +1,21 @@
package state package state
import ( import (
"fmt"
"testing" "testing"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
abci "github.com/tendermint/abci/types" abci "github.com/tendermint/abci/types"
crypto "github.com/tendermint/go-crypto" crypto "github.com/tendermint/go-crypto"
cfg "github.com/tendermint/tendermint/config"
cmn "github.com/tendermint/tmlibs/common"
dbm "github.com/tendermint/tmlibs/db" dbm "github.com/tendermint/tmlibs/db"
"github.com/tendermint/tmlibs/log" "github.com/tendermint/tmlibs/log"
cfg "github.com/tendermint/tendermint/config"
) )
func TestStateCopyEquals(t *testing.T) { func TestStateCopyEquals(t *testing.T) {
assert := assert.New(t)
config := cfg.ResetTestRoot("state_") config := cfg.ResetTestRoot("state_")
// Get State db // Get State db
@ -22,18 +25,13 @@ func TestStateCopyEquals(t *testing.T) {
stateCopy := state.Copy() stateCopy := state.Copy()
if !state.Equals(stateCopy) {
t.Fatal("expected state and its copy to be identical. got %v\n expected %v\n", stateCopy, state)
}
assert.True(state.Equals(stateCopy), cmn.Fmt("expected state and its copy to be identical. got %v\n expected %v\n", stateCopy, state))
stateCopy.LastBlockHeight += 1 stateCopy.LastBlockHeight += 1
if state.Equals(stateCopy) {
t.Fatal("expected states to be different. got same %v", state)
}
assert.False(state.Equals(stateCopy), cmn.Fmt("expected states to be different. got same %v", state))
} }
func TestStateSaveLoad(t *testing.T) { func TestStateSaveLoad(t *testing.T) {
assert := assert.New(t)
config := cfg.ResetTestRoot("state_") config := cfg.ResetTestRoot("state_")
// Get State db // Get State db
stateDB := dbm.NewDB("state", config.DBBackend, config.DBDir()) stateDB := dbm.NewDB("state", config.DBBackend, config.DBDir())
@ -44,9 +42,7 @@ func TestStateSaveLoad(t *testing.T) {
state.Save() state.Save()
loadedState := LoadState(stateDB) loadedState := LoadState(stateDB)
if !state.Equals(loadedState) {
t.Fatal("expected state and its copy to be identical. got %v\n expected %v\n", loadedState, state)
}
assert.True(state.Equals(loadedState), cmn.Fmt("expected state and its copy to be identical. got %v\n expected %v\n", loadedState, state))
} }
func TestABCIResponsesSaveLoad(t *testing.T) { func TestABCIResponsesSaveLoad(t *testing.T) {
@ -74,5 +70,41 @@ func TestABCIResponsesSaveLoad(t *testing.T) {
state.SaveABCIResponses(abciResponses) state.SaveABCIResponses(abciResponses)
abciResponses2 := state.LoadABCIResponses() abciResponses2 := state.LoadABCIResponses()
assert.Equal(abciResponses, abciResponses2, fmt.Sprintf("ABCIResponses don't match: Got %v, Expected %v", abciResponses2, abciResponses))
assert.Equal(abciResponses, abciResponses2, cmn.Fmt("ABCIResponses don't match: Got %v, Expected %v", abciResponses2, abciResponses))
}
func TestValidatorsSaveLoad(t *testing.T) {
assert := assert.New(t)
config := cfg.ResetTestRoot("state_")
// Get State db
stateDB := dbm.NewDB("state", config.DBBackend, config.DBDir())
state := GetState(stateDB, config.GenesisFile())
state.SetLogger(log.TestingLogger())
// cant load anything for height 0
v, err := state.LoadValidators(0)
assert.NotNil(err, "expected err at height 0")
// should be able to load for height 1
v, err = state.LoadValidators(1)
assert.Nil(err, "expected no err at height 1")
assert.Equal(v.Hash(), state.Validators.Hash(), "expected validator hashes to match")
// increment height, save; should be able to load for next height
state.LastBlockHeight += 1
state.SaveValidators()
v, err = state.LoadValidators(state.LastBlockHeight + 1)
assert.Nil(err, "expected no err")
assert.Equal(v.Hash(), state.Validators.Hash(), "expected validator hashes to match")
// increment height, save; should be able to load for next height
state.LastBlockHeight += 10
state.SaveValidators()
v, err = state.LoadValidators(state.LastBlockHeight + 1)
assert.Nil(err, "expected no err")
assert.Equal(v.Hash(), state.Validators.Hash(), "expected validator hashes to match")
// should be able to load for next next height
_, err = state.LoadValidators(state.LastBlockHeight + 2)
assert.NotNil(err, "expected err")
} }

Loading…
Cancel
Save