fix EvidencePool and VerifyEvidence

evidence/pool.go

@@ -1,6 +1,10 @@
 package evidence
 
 import (
+	"fmt"
+	"sync"
+
+	dbm "github.com/tendermint/tmlibs/db"
 	"github.com/tendermint/tmlibs/log"
 
 	sm "github.com/tendermint/tendermint/state"
@@ -14,17 +18,21 @@ type EvidencePool struct {
 
 	evidenceStore *EvidenceStore
 
-	state  sm.State
-	params types.EvidenceParams
+	// needed to load validators to verify evidence
+	stateDB dbm.DB
+
+	// latest state
+	mtx   sync.Mutex
+	state sm.State
 
 	// never close
 	evidenceChan chan types.Evidence
 }
 
-func NewEvidencePool(params types.EvidenceParams, evidenceStore *EvidenceStore, state sm.State) *EvidencePool {
+func NewEvidencePool(stateDB dbm.DB, evidenceStore *EvidenceStore) *EvidencePool {
 	evpool := &EvidencePool{
-		state:         state,
-		params:        params,
+		stateDB:       stateDB,
+		state:         sm.LoadState(stateDB),
 		logger:        log.NewNopLogger(),
 		evidenceStore: evidenceStore,
 		evidenceChan:  make(chan types.Evidence),
@@ -52,31 +60,44 @@ func (evpool *EvidencePool) PendingEvidence() []types.Evidence {
 	return evpool.evidenceStore.PendingEvidence()
 }
 
+// State returns the current state of the evpool.
+func (evpool *EvidencePool) State() sm.State {
+	evpool.mtx.Lock()
+	defer evpool.mtx.Unlock()
+	return evpool.state
+}
+
+// Update loads the latest
+func (evpool *EvidencePool) Update(block *types.Block) {
+	evpool.mtx.Lock()
+	defer evpool.mtx.Unlock()
+
+	state := sm.LoadState(evpool.stateDB)
+	if state.LastBlockHeight != block.Height {
+		panic(fmt.Sprintf("EvidencePool.Update: loaded state with height %d when block.Height=%d", state.LastBlockHeight, block.Height))
+	}
+	evpool.state = state
+
+	// NOTE: shouldn't need the mutex
+	evpool.MarkEvidenceAsCommitted(block.Evidence.Evidence)
+}
+
 // AddEvidence checks the evidence is valid and adds it to the pool.
 // Blocks on the EvidenceChan.
 func (evpool *EvidencePool) AddEvidence(evidence types.Evidence) (err error) {
+
 	// TODO: check if we already have evidence for this
 	// validator at this height so we dont get spammed
 
-	if err := sm.VerifyEvidence(evpool.state, evidence); err != nil {
+	if err := sm.VerifyEvidence(evpool.stateDB, evpool.State(), evidence); err != nil {
 		return err
 	}
 
-	var priority int64
-	/* // Needs a db ...
-	// TODO: if err is just that we cant find it cuz we pruned, ignore.
-	// TODO: if its actually bad evidence, punish peer
-
-	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)
-	}
-	*/
+	// fetch the validator and return its voting power as its priority
+	// TODO: something better ?
+	valset, _ := sm.LoadValidators(evpool.stateDB, evidence.Height())
+	_, val := valset.GetByAddress(evidence.Address())
+	priority := val.VotingPower
 
 	added := evpool.evidenceStore.AddNewEvidence(evidence, priority)
 	if !added {

evidence/pool_test.go

@@ -3,6 +3,7 @@ package evidence
 import (
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 
@@ -13,14 +14,46 @@ import (
 
 var mockState = sm.State{}
 
+func initializeValidatorState(valAddr []byte, height int64) dbm.DB {
+	stateDB := dbm.NewMemDB()
+
+	// create validator set and state
+	valSet := &types.ValidatorSet{
+		Validators: []*types.Validator{
+			{Address: valAddr},
+		},
+	}
+	state := sm.State{
+		LastBlockHeight:             0,
+		LastBlockTime:               time.Now(),
+		Validators:                  valSet,
+		LastHeightValidatorsChanged: 1,
+		ConsensusParams: types.ConsensusParams{
+			EvidenceParams: types.EvidenceParams{
+				MaxAge: 1000000,
+			},
+		},
+	}
+
+	// save all states up to height
+	for i := int64(0); i < height; i++ {
+		state.LastBlockHeight = i
+		sm.SaveState(stateDB, state)
+	}
+
+	return stateDB
+}
+
 func TestEvidencePool(t *testing.T) {
 	assert := assert.New(t)
 
-	params := types.EvidenceParams{}
+	valAddr := []byte("val1")
+	height := int64(5)
+	stateDB := initializeValidatorState(valAddr, height)
 	store := NewEvidenceStore(dbm.NewMemDB())
-	pool := NewEvidencePool(params, store, mockState)
+	pool := NewEvidencePool(stateDB, store)
 
-	goodEvidence := newMockGoodEvidence(5, 1, []byte("val1"))
+	goodEvidence := newMockGoodEvidence(height, 0, valAddr)
 	badEvidence := MockBadEvidence{goodEvidence}
 
 	err := pool.AddEvidence(badEvidence)

evidence/reactor_test.go

@@ -32,14 +32,14 @@ func evidenceLogger() log.Logger {
 }
 
 // connect N evidence reactors through N switches
-func makeAndConnectEvidenceReactors(config *cfg.Config, N int) []*EvidenceReactor {
+func makeAndConnectEvidenceReactors(config *cfg.Config, stateDBs []dbm.DB) []*EvidenceReactor {
+	N := len(stateDBs)
 	reactors := make([]*EvidenceReactor, N)
 	logger := evidenceLogger()
 	for i := 0; i < N; i++ {
 
-		params := types.EvidenceParams{}
 		store := NewEvidenceStore(dbm.NewMemDB())
-		pool := NewEvidencePool(params, store, mockState)
+		pool := NewEvidencePool(stateDBs[i], store)
 		reactors[i] = NewEvidenceReactor(pool)
 		reactors[i].SetLogger(logger.With("validator", i))
 	}
@@ -98,10 +98,10 @@ func _waitForEvidence(t *testing.T, wg *sync.WaitGroup, evs types.EvidenceList,
 	wg.Done()
 }
 
-func sendEvidence(t *testing.T, evpool *EvidencePool, n int) types.EvidenceList {
+func sendEvidence(t *testing.T, evpool *EvidencePool, valAddr []byte, n int) types.EvidenceList {
 	evList := make([]types.Evidence, n)
 	for i := 0; i < n; i++ {
-		ev := newMockGoodEvidence(int64(i), 2, []byte("val"))
+		ev := newMockGoodEvidence(int64(i+1), 0, valAddr)
 		err := evpool.AddEvidence(ev)
 		assert.Nil(t, err)
 		evList[i] = ev
@@ -110,17 +110,28 @@ func sendEvidence(t *testing.T, evpool *EvidencePool, n int) types.EvidenceList
 }
 
 var (
-	NUM_EVIDENCE = 1000
+	NUM_EVIDENCE = 1
 	TIMEOUT      = 120 * time.Second // ridiculously high because CircleCI is slow
 )
 
 func TestReactorBroadcastEvidence(t *testing.T) {
 	config := cfg.TestConfig()
 	N := 7
-	reactors := makeAndConnectEvidenceReactors(config, N)
 
-	// send a bunch of evidence to the first reactor's evpool
+	// create statedb for everyone
+	stateDBs := make([]dbm.DB, N)
+	valAddr := []byte("myval")
+	// we need validators saved for heights at least as high as we have evidence for
+	height := int64(NUM_EVIDENCE) + 10
+	for i := 0; i < N; i++ {
+		stateDBs[i] = initializeValidatorState(valAddr, height)
+	}
+
+	// make reactors from statedb
+	reactors := makeAndConnectEvidenceReactors(config, stateDBs)
+
+	// send a bunch of valid evidence to the first reactor's evpool
 	// and wait for them all to be received in the others
-	evList := sendEvidence(t, reactors[0].evpool, NUM_EVIDENCE)
+	evList := sendEvidence(t, reactors[0].evpool, valAddr, NUM_EVIDENCE)
 	waitForEvidence(t, evList, reactors)
 }

node/node.go

@@ -208,7 +208,7 @@ func NewNode(config *cfg.Config,
 	}
 	evidenceLogger := logger.With("module", "evidence")
 	evidenceStore := evidence.NewEvidenceStore(evidenceDB)
-	evidencePool := evidence.NewEvidencePool(state.ConsensusParams.EvidenceParams, evidenceStore, state.Copy())
+	evidencePool := evidence.NewEvidencePool(stateDB, evidenceStore)
 	evidencePool.SetLogger(evidenceLogger)
 	evidenceReactor := evidence.NewEvidenceReactor(evidencePool)
 	evidenceReactor.SetLogger(evidenceLogger)

state/execution.go

@@ -107,6 +107,11 @@ func (blockExec *BlockExecutor) ApplyBlock(s State, blockID types.BlockID, block
 
 	fail.Fail() // XXX
 
+	// Update evpool now that state is saved
+	// TODO: handle the crash/recover scenario
+	// ie. (may need to call Update for last block)
+	blockExec.evpool.Update(block)
+
 	// events are fired after everything else
 	// NOTE: if we crash between Commit and Save, events wont be fired during replay
 	fireEvents(blockExec.logger, blockExec.eventBus, block, abciResponses)
@@ -138,9 +143,6 @@ func (blockExec *BlockExecutor) Commit(block *types.Block) ([]byte, error) {
 
 	blockExec.logger.Info("Committed state", "height", block.Height, "txs", block.NumTxs, "appHash", 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

state/validation.go

@@ -79,20 +79,9 @@ func validateBlock(stateDB dbm.DB, s State, b *types.Block) error {
 	}
 
 	for _, ev := range b.Evidence.Evidence {
-		if err := VerifyEvidence(s, ev); err != nil {
+		if err := VerifyEvidence(stateDB, 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
@@ -103,7 +92,7 @@ func validateBlock(stateDB dbm.DB, s State, b *types.Block) error {
 
 // VerifyEvidence verifies the evidence fully by checking it is internally
 // consistent and sufficiently recent.
-func VerifyEvidence(s State, evidence types.Evidence) error {
+func VerifyEvidence(stateDB dbm.DB, s State, evidence types.Evidence) error {
 	height := s.LastBlockHeight
 
 	evidenceAge := height - evidence.Height()
@@ -116,22 +105,23 @@ func VerifyEvidence(s State, evidence types.Evidence) error {
 	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) {
+	valset, err := LoadValidators(stateDB, evidence.Height())
+	if err != nil {
+		// TODO: if err is just that we cant find it cuz we pruned, ignore.
+		// TODO: if its actually bad evidence, punish peer
+		return err
+	}
+
 	// 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)
+		return 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)
+		return fmt.Errorf("Address %X was validator %d at height %d, not %d", addr, valIdx, height, idx)
 	}
 
-	priority = val.VotingPower
-	return priority, nil
+	return nil
 }

state/validation_test.go

@@ -4,61 +4,65 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/require"
+	dbm "github.com/tendermint/tmlibs/db"
+	"github.com/tendermint/tmlibs/log"
 )
 
-func _TestValidateBlock(t *testing.T) {
+func TestValidateBlock(t *testing.T) {
 	state := state()
 
+	blockExec := NewBlockExecutor(dbm.NewMemDB(), log.TestingLogger(), nil, nil, nil)
+
 	// proper block must pass
 	block := makeBlock(state, 1)
-	err := ValidateBlock(state, block)
+	err := blockExec.ValidateBlock(state, block)
 	require.NoError(t, err)
 
 	// wrong chain fails
 	block = makeBlock(state, 1)
 	block.ChainID = "not-the-real-one"
-	err = ValidateBlock(state, block)
+	err = blockExec.ValidateBlock(state, block)
 	require.Error(t, err)
 
 	// wrong height fails
 	block = makeBlock(state, 1)
 	block.Height += 10
-	err = ValidateBlock(state, block)
+	err = blockExec.ValidateBlock(state, block)
 	require.Error(t, err)
 
 	// wrong total tx fails
 	block = makeBlock(state, 1)
 	block.TotalTxs += 10
-	err = ValidateBlock(state, block)
+	err = blockExec.ValidateBlock(state, block)
 	require.Error(t, err)
 
 	// wrong blockid fails
 	block = makeBlock(state, 1)
 	block.LastBlockID.PartsHeader.Total += 10
-	err = ValidateBlock(state, block)
+	err = blockExec.ValidateBlock(state, block)
 	require.Error(t, err)
 
 	// wrong app hash fails
 	block = makeBlock(state, 1)
 	block.AppHash = []byte("wrong app hash")
-	err = ValidateBlock(state, block)
+	err = blockExec.ValidateBlock(state, block)
 	require.Error(t, err)
 
 	// wrong consensus hash fails
 	block = makeBlock(state, 1)
 	block.ConsensusHash = []byte("wrong consensus hash")
-	err = ValidateBlock(state, block)
+	err = blockExec.ValidateBlock(state, block)
 	require.Error(t, err)
 
 	// wrong results hash fails
 	block = makeBlock(state, 1)
 	block.LastResultsHash = []byte("wrong results hash")
-	err = ValidateBlock(state, block)
+	err = blockExec.ValidateBlock(state, block)
 	require.Error(t, err)
 
 	// wrong validators hash fails
 	block = makeBlock(state, 1)
 	block.ValidatorsHash = []byte("wrong validators hash")
-	err = ValidateBlock(state, block)
+	err = blockExec.ValidateBlock(state, block)
 	require.Error(t, err)
 }

types/services.go

@@ -78,7 +78,7 @@ type BlockStore interface {
 type EvidencePool interface {
 	PendingEvidence() []Evidence
 	AddEvidence(Evidence) error
-	MarkEvidenceAsCommitted([]Evidence)
+	Update(*Block)
 }
 
 // MockMempool is an empty implementation of a Mempool, useful for testing.
@@ -86,6 +86,6 @@ type EvidencePool interface {
 type MockEvidencePool struct {
 }
 
-func (m MockEvidencePool) PendingEvidence() []Evidence        { return nil }
-func (m MockEvidencePool) AddEvidence(Evidence) error         { return nil }
-func (m MockEvidencePool) MarkEvidenceAsCommitted([]Evidence) {}
+func (m MockEvidencePool) PendingEvidence() []Evidence { return nil }
+func (m MockEvidencePool) AddEvidence(Evidence) error  { return nil }
+func (m MockEvidencePool) Update(*Block)               {}