Minimalistic validator set handling (cosmos#286)

For Tendermint/Cosmos-SDK compatibility, validator set handling needs to be supported. Optimint's aggregators are analogues of Tendermint validators. Because of that, validator set handling is the most simple, Cosmos-SDK compatible form of managing aggregators.

CHANGELOG-PENDING.md

@@ -27,10 +27,11 @@ Month, DD, YYYY
 - [rpc] [Implement Commit and BlockSearch method #258](https://github.com/celestiaorg/optimint/pull/258) [@raneet10](https://github.com/Raneet10/)
 - [rpc] [Remove extra variable #280](https://github.com/celestiaorg/optimint/pull/280) [@raneet10](https://github.com/Raneet10/)
 - [rpc] [Implement BlockChainInfo RPC method #282](https://github.com/celestiaorg/optimint/pull/282) [@raneet10](https://github.com/Raneet10/)
+- [state,block,store,rpc] [Minimalistic validator set handling](https://github.com/celestiaorg/optimint/pull/286) [@tzdybal](https://github.com/tzdybal/)
 
 ### BUG FIXES
 
 - [store] [Use KeyCopy instead of Key in BadgerIterator #274](https://github.com/celestiaorg/optimint/pull/274) [@tzdybal](https://github.com/tzdybal/)
-- [state/block] [Do save ABCI responses for blocks #285r](https://github.com/celestiaorg/optimint/pull/285) [@tzdybal](https://github.com/tzdybal/)
+- [state,block] [Do save ABCI responses for blocks #285r](https://github.com/celestiaorg/optimint/pull/285) [@tzdybal](https://github.com/tzdybal/)
 
 - [go package] (Link to PR) Description @username

block/manager.go

@@ -311,6 +311,11 @@ func (m *Manager) publishBlock(ctx context.Context) error {
 		return err
 	}
 
+	err = m.store.SaveValidators(block.Header.Height, m.lastState.Validators)
+	if err != nil {
+		return err
+	}
+
 	return m.broadcastBlock(ctx, block)
 }
 

rpc/client/client.go

@@ -433,8 +433,28 @@ func (c *Client) Commit(ctx context.Context, height *int64) (*ctypes.ResultCommi
 	return ctypes.NewResultCommit(&block.Header, commit, true), nil
 }
 
-func (c *Client) Validators(ctx context.Context, height *int64, page, perPage *int) (*ctypes.ResultValidators, error) {
-	panic("Validators - not implemented!")
+func (c *Client) Validators(ctx context.Context, heightPtr *int64, pagePtr, perPagePtr *int) (*ctypes.ResultValidators, error) {
+	height := c.normalizeHeight(heightPtr)
+	validators, err := c.node.Store.LoadValidators(height)
+	if err != nil {
+		return nil, fmt.Errorf("failed to load validators for height %d: %w", height, err)
+	}
+
+	totalCount := len(validators.Validators)
+	perPage := validatePerPage(perPagePtr)
+	page, err := validatePage(pagePtr, perPage, totalCount)
+	if err != nil {
+		return nil, err
+	}
+
+	skipCount := validateSkipCount(page, perPage)
+	v := validators.Validators[skipCount : skipCount+tmmath.MinInt(perPage, totalCount-skipCount)]
+	return &ctypes.ResultValidators{
+		BlockHeight: int64(height),
+		Validators:  v,
+		Count:       len(v),
+		Total:       totalCount,
+	}, nil
 }
 
 func (c *Client) Tx(ctx context.Context, hash []byte, prove bool) (*ctypes.ResultTx, error) {

rpc/client/client_test.go

@@ -15,6 +15,9 @@ import (
 	"github.com/libp2p/go-libp2p-core/crypto"
 	"github.com/libp2p/go-libp2p-core/peer"
 	abci "github.com/tendermint/tendermint/abci/types"
+	tmcrypto "github.com/tendermint/tendermint/crypto"
+	"github.com/tendermint/tendermint/crypto/ed25519"
+	"github.com/tendermint/tendermint/crypto/encoding"
 	"github.com/tendermint/tendermint/libs/bytes"
 	"github.com/tendermint/tendermint/libs/log"
 	tmproto "github.com/tendermint/tendermint/proto/tendermint/types"
@@ -572,6 +575,78 @@ func TestBlockchainInfo(t *testing.T) {
 	}
 }
 
+func TestValidatorSetHandling(t *testing.T) {
+	assert := assert.New(t)
+	require := require.New(t)
+	app := &mocks.Application{}
+	app.On("InitChain", mock.Anything).Return(abci.ResponseInitChain{})
+	app.On("CheckTx", mock.Anything).Return(abci.ResponseCheckTx{})
+	app.On("BeginBlock", mock.Anything).Return(abci.ResponseBeginBlock{})
+	app.On("Commit", mock.Anything).Return(abci.ResponseCommit{})
+
+	key, _, _ := crypto.GenerateEd25519Key(crand.Reader)
+
+	vKeys := make([]tmcrypto.PrivKey, 4)
+	genesisValidators := make([]tmtypes.GenesisValidator, len(vKeys))
+	for i := 0; i < len(vKeys); i++ {
+		vKeys[i] = ed25519.GenPrivKey()
+		genesisValidators[i] = tmtypes.GenesisValidator{Address: vKeys[i].PubKey().Address(), PubKey: vKeys[i].PubKey(), Power: int64(i + 100), Name: "one"}
+	}
+
+	pbValKey, err := encoding.PubKeyToProto(vKeys[0].PubKey())
+	require.NoError(err)
+
+	app.On("EndBlock", mock.Anything).Return(abci.ResponseEndBlock{}).Times(5)
+	app.On("EndBlock", mock.Anything).Return(abci.ResponseEndBlock{ValidatorUpdates: []abci.ValidatorUpdate{{PubKey: pbValKey, Power: 0}}}).Once()
+	app.On("EndBlock", mock.Anything).Return(abci.ResponseEndBlock{}).Once()
+	app.On("EndBlock", mock.Anything).Return(abci.ResponseEndBlock{ValidatorUpdates: []abci.ValidatorUpdate{{PubKey: pbValKey, Power: 100}}}).Once()
+	app.On("EndBlock", mock.Anything).Return(abci.ResponseEndBlock{})
+
+	node, err := node.NewNode(context.Background(), config.NodeConfig{DALayer: "mock", Aggregator: true, BlockManagerConfig: config.BlockManagerConfig{BlockTime: 10 * time.Millisecond}}, key, proxy.NewLocalClientCreator(app), &tmtypes.GenesisDoc{ChainID: "test", Validators: genesisValidators}, log.TestingLogger())
+	require.NoError(err)
+	require.NotNil(node)
+
+	rpc := NewClient(node)
+	require.NotNil(rpc)
+
+	err = node.Start()
+	require.NoError(err)
+
+	// test latest block a few times - ensure that validator set from genesis is handled correctly
+	lastHeight := int64(-1)
+	for i := 0; i < 3; i++ {
+		time.Sleep(10 * time.Millisecond)
+		vals, err := rpc.Validators(context.Background(), nil, nil, nil)
+		assert.NoError(err)
+		assert.NotNil(vals)
+		assert.EqualValues(len(genesisValidators), vals.Total)
+		assert.Len(vals.Validators, len(genesisValidators))
+		assert.Greater(vals.BlockHeight, lastHeight)
+		lastHeight = vals.BlockHeight
+	}
+	time.Sleep(100 * time.Millisecond)
+
+	// 6th EndBlock removes first validator from the list
+	for h := int64(7); h <= 8; h++ {
+		vals, err := rpc.Validators(context.Background(), &h, nil, nil)
+		assert.NoError(err)
+		assert.NotNil(vals)
+		assert.EqualValues(len(genesisValidators)-1, vals.Total)
+		assert.Len(vals.Validators, len(genesisValidators)-1)
+		assert.EqualValues(vals.BlockHeight, h)
+	}
+
+	// 8th EndBlock adds validator back
+	for h := int64(9); h < 12; h++ {
+		vals, err := rpc.Validators(context.Background(), &h, nil, nil)
+		assert.NoError(err)
+		assert.NotNil(vals)
+		assert.EqualValues(len(genesisValidators), vals.Total)
+		assert.Len(vals.Validators, len(genesisValidators))
+		assert.EqualValues(vals.BlockHeight, h)
+	}
+}
+
 // copy-pasted from store/store_test.go
 func getRandomBlock(height uint64, nTxs int) *types.Block {
 	block := &types.Block{

state/executor.go

@@ -4,9 +4,11 @@ import (
 	"bytes"
 	"context"
 	"errors"
+	"fmt"
 	"time"
 
 	abci "github.com/tendermint/tendermint/abci/types"
+	cryptoenc "github.com/tendermint/tendermint/crypto/encoding"
 	tmstate "github.com/tendermint/tendermint/proto/tendermint/state"
 	tmproto "github.com/tendermint/tendermint/proto/tendermint/types"
 	"github.com/tendermint/tendermint/proxy"
@@ -48,6 +50,12 @@ func NewBlockExecutor(proposerAddress []byte, namespaceID [8]byte, chainID strin
 
 func (e *BlockExecutor) InitChain(genesis *tmtypes.GenesisDoc) (*abci.ResponseInitChain, error) {
 	params := genesis.ConsensusParams
+
+	validators := make([]*tmtypes.Validator, len(genesis.Validators))
+	for i, v := range genesis.Validators {
+		validators[i] = tmtypes.NewValidator(v.PubKey, v.Power)
+	}
+
 	return e.proxyApp.InitChainSync(abci.RequestInitChain{
 		Time:    genesis.GenesisTime,
 		ChainId: genesis.ChainID,
@@ -68,7 +76,7 @@ func (e *BlockExecutor) InitChain(genesis *tmtypes.GenesisDoc) (*abci.ResponseIn
 				AppVersion: params.Version.AppVersion,
 			},
 		},
-		Validators:    nil,
+		Validators:    tmtypes.TM2PB.ValidatorUpdates(tmtypes.NewValidatorSet(validators)),
 		AppStateBytes: genesis.AppState,
 		InitialHeight: genesis.InitialHeight,
 	})
@@ -121,7 +129,21 @@ func (e *BlockExecutor) ApplyBlock(ctx context.Context, state State, block *type
 		return State{}, nil, 0, err
 	}
 
-	state, err = e.updateState(state, block, resp)
+	abciValUpdates := resp.EndBlock.ValidatorUpdates
+	err = validateValidatorUpdates(abciValUpdates, state.ConsensusParams.Validator)
+	if err != nil {
+		return state, nil, 0, fmt.Errorf("error in validator updates: %v", err)
+	}
+
+	validatorUpdates, err := tmtypes.PB2TM.ValidatorUpdates(abciValUpdates)
+	if err != nil {
+		return state, nil, 0, err
+	}
+	if len(validatorUpdates) > 0 {
+		e.logger.Debug("updates to validators", "updates", tmtypes.ValidatorListString(validatorUpdates))
+	}
+
+	state, err = e.updateState(state, block, resp, validatorUpdates)
 	if err != nil {
 		return State{}, nil, 0, err
 	}
@@ -141,7 +163,24 @@ func (e *BlockExecutor) ApplyBlock(ctx context.Context, state State, block *type
 	return state, resp, retainHeight, nil
 }
 
-func (e *BlockExecutor) updateState(state State, block *types.Block, abciResponses *tmstate.ABCIResponses) (State, error) {
+func (e *BlockExecutor) updateState(state State, block *types.Block, abciResponses *tmstate.ABCIResponses, validatorUpdates []*tmtypes.Validator) (State, error) {
+	nValSet := state.NextValidators.Copy()
+	lastHeightValSetChanged := state.LastHeightValidatorsChanged
+	// Optimint can work without validators
+	if len(nValSet.Validators) > 0 {
+		if len(validatorUpdates) > 0 {
+			err := nValSet.UpdateWithChangeSet(validatorUpdates)
+			if err != nil {
+				return state, nil
+			}
+			// Change results from this height but only applies to the next next height.
+			lastHeightValSetChanged = int64(block.Header.Height + 1 + 1)
+		}
+
+		// TODO(tzdybal):  right now, it's for backward compatibility, may need to change this
+		nValSet.IncrementProposerPriority(1)
+	}
+
 	hash := block.Header.Hash()
 	s := State{
 		Version:         state.Version,
@@ -153,9 +192,13 @@ func (e *BlockExecutor) updateState(state State, block *types.Block, abciRespons
 			Hash: hash[:],
 			// for now, we don't care about part set headers
 		},
-		// skipped all "Validators" fields
+		NextValidators:                   nValSet,
+		Validators:                       state.NextValidators.Copy(),
+		LastValidators:                   state.Validators.Copy(),
+		LastHeightValidatorsChanged:      lastHeightValSetChanged,
 		ConsensusParams:                  state.ConsensusParams,
 		LastHeightConsensusParamsChanged: state.LastHeightConsensusParamsChanged,
+		AppHash:                          [32]byte{},
 	}
 	copy(s.LastResultsHash[:], tmtypes.NewResults(abciResponses.DeliverTxs).Hash())
 
@@ -335,3 +378,28 @@ func fromOptimintTxs(optiTxs types.Txs) tmtypes.Txs {
 	}
 	return txs
 }
+
+func validateValidatorUpdates(abciUpdates []abci.ValidatorUpdate,
+	params tmproto.ValidatorParams) error {
+	for _, valUpdate := range abciUpdates {
+		if valUpdate.GetPower() < 0 {
+			return fmt.Errorf("voting power can't be negative %v", valUpdate)
+		} else if valUpdate.GetPower() == 0 {
+			// continue, since this is deleting the validator, and thus there is no
+			// pubkey to check
+			continue
+		}
+
+		// Check if validator's pubkey matches an ABCI type in the consensus params
+		pk, err := cryptoenc.PubKeyFromProto(valUpdate.PubKey)
+		if err != nil {
+			return err
+		}
+
+		if !tmtypes.IsValidPubkeyType(params, pk.Type()) {
+			return fmt.Errorf("validator %v is using pubkey %s, which is unsupported for consensus",
+				valUpdate, pk.Type())
+		}
+	}
+	return nil
+}

state/executor_test.go

@@ -110,7 +110,11 @@ func TestApplyBlock(t *testing.T) {
 	require.NoError(err)
 	require.NotNil(headerSub)
 
-	state := State{}
+	state := State{
+		NextValidators: tmtypes.NewValidatorSet(nil),
+		Validators:     tmtypes.NewValidatorSet(nil),
+		LastValidators: tmtypes.NewValidatorSet(nil),
+	}
 	state.InitialHeight = 1
 	state.LastBlockHeight = 0
 	state.ConsensusParams.Block.MaxBytes = 100

store/store.go

@@ -7,18 +7,21 @@ import (
 	"sync"
 
 	tmstate "github.com/tendermint/tendermint/proto/tendermint/state"
+	tmproto "github.com/tendermint/tendermint/proto/tendermint/types"
+	tmtypes "github.com/tendermint/tendermint/types"
 	"go.uber.org/multierr"
 
 	"github.com/celestiaorg/optimint/state"
 	"github.com/celestiaorg/optimint/types"
 )
 
 var (
-	blockPrefix     = [1]byte{1}
-	indexPrefix     = [1]byte{2}
-	commitPrefix    = [1]byte{3}
-	statePrefix     = [1]byte{4}
-	responsesPrefix = [1]byte{5}
+	blockPrefix      = [1]byte{1}
+	indexPrefix      = [1]byte{2}
+	commitPrefix     = [1]byte{3}
+	statePrefix      = [1]byte{4}
+	responsesPrefix  = [1]byte{5}
+	validatorsPrefix = [1]byte{6}
 )
 
 // DefaultStore is a default store implmementation.
@@ -177,6 +180,33 @@ func (s *DefaultStore) LoadState() (state.State, error) {
 	return state, err
 }
 
+func (s *DefaultStore) SaveValidators(height uint64, validatorSet *tmtypes.ValidatorSet) error {
+	pbValSet, err := validatorSet.ToProto()
+	if err != nil {
+		return err
+	}
+	blob, err := pbValSet.Marshal()
+	if err != nil {
+		return err
+	}
+
+	return s.db.Set(getValidatorsKey(height), blob)
+}
+
+func (s *DefaultStore) LoadValidators(height uint64) (*tmtypes.ValidatorSet, error) {
+	blob, err := s.db.Get(getValidatorsKey(height))
+	if err != nil {
+		return nil, err
+	}
+	var pbValSet tmproto.ValidatorSet
+	err = pbValSet.Unmarshal(blob)
+	if err != nil {
+		return nil, err
+	}
+
+	return tmtypes.ValidatorSetFromProto(&pbValSet)
+}
+
 func (s *DefaultStore) loadHashFromIndex(height uint64) ([32]byte, error) {
 	blob, err := s.db.Get(getIndexKey(height))
 
@@ -214,3 +244,9 @@ func getResponsesKey(height uint64) []byte {
 	binary.BigEndian.PutUint64(buf, height)
 	return append(responsesPrefix[:], buf[:]...)
 }
+
+func getValidatorsKey(height uint64) []byte {
+	buf := make([]byte, 8)
+	binary.BigEndian.PutUint64(buf, height)
+	return append(validatorsPrefix[:], buf[:]...)
+}

store/types.go

@@ -2,6 +2,7 @@ package store
 
 import (
 	tmstate "github.com/tendermint/tendermint/proto/tendermint/state"
+	tmtypes "github.com/tendermint/tendermint/types"
 
 	"github.com/celestiaorg/optimint/state"
 	"github.com/celestiaorg/optimint/types"
@@ -36,4 +37,8 @@ type Store interface {
 	UpdateState(state state.State) error
 	// LoadState returns last state saved with UpdateState.
 	LoadState() (state.State, error)
+
+	SaveValidators(height uint64, validatorSet *tmtypes.ValidatorSet) error
+
+	LoadValidators(height uint64) (*tmtypes.ValidatorSet, error)
 }