check evidence is from validator; some cleanup

evidence/evidence_pool.go

@@ -42,20 +42,17 @@ func (evpool *EvidencePool) NewEvidenceChan() chan types.Evidence {
 
 // PriorityEvidence returns the priority evidence.
 func (evpool *EvidencePool) PriorityEvidence() []types.Evidence {
-	// TODO
-	return nil
+	return evpool.evidenceStore.PriorityEvidence()
 }
 
 // PendingEvidence returns all uncommitted evidence.
 func (evpool *EvidencePool) PendingEvidence() []types.Evidence {
-	// TODO
-	return nil
+	return evpool.evidenceStore.PendingEvidence()
 }
 
 // AddEvidence checks the evidence is valid and adds it to the pool.
 func (evpool *EvidencePool) AddEvidence(evidence types.Evidence) (err error) {
-	idx := 1 // TODO
-	added, err := evpool.evidenceStore.AddNewEvidence(idx, evidence)
+	added, err := evpool.evidenceStore.AddNewEvidence(evidence)
 	if err != nil {
 		return err
 	} else if !added {

evidence/reactor.go

@@ -67,20 +67,19 @@ func (evR *EvidencePoolReactor) GetChannels() []*p2p.ChannelDescriptor {
 
 // AddPeer implements Reactor.
 func (evR *EvidencePoolReactor) AddPeer(peer p2p.Peer) {
-	// first send the peer high-priority evidence
+	// send the peer our high-priority evidence.
+	// the rest will be sent by the broadcastRoutine
 	evidence := evR.evpool.PriorityEvidence()
 	msg := EvidenceMessage{evidence}
 	success := peer.Send(EvidencePoolChannel, struct{ EvidencePoolMessage }{msg})
 	if !success {
 		// TODO: remove peer ?
 	}
-
-	// TODO: send the remaining pending evidence
-	// or just let the broadcastRoutine do it ?
 }
 
 // RemovePeer implements Reactor.
 func (evR *EvidencePoolReactor) RemovePeer(peer p2p.Peer, reason interface{}) {
+	// nothing to do
 }
 
 // Receive implements Reactor.
@@ -122,10 +121,9 @@ func (evR *EvidencePoolReactor) broadcastRoutine() {
 			msg := EvidenceMessage{[]types.Evidence{evidence}}
 			evR.Switch.Broadcast(EvidencePoolChannel, struct{ EvidencePoolMessage }{msg})
 
-			// NOTE: Broadcast runs asynchronously, so this should wait on the successChan
+			// TODO: Broadcast runs asynchronously, so this should wait on the successChan
 			// in another routine before marking to be proper.
-			idx := 1 // TODO
-			evR.evpool.evidenceStore.MarkEvidenceAsBroadcasted(idx, evidence)
+			evR.evpool.evidenceStore.MarkEvidenceAsBroadcasted(evidence)
 		case <-ticker.C:
 			// broadcast all pending evidence
 			msg := EvidenceMessage{evR.evpool.PendingEvidence()}

evidence/store.go

@@ -22,16 +22,22 @@ type evidenceInfo struct {
 	Evidence  types.Evidence
 }
 
+const (
+	baseKeyLookup   = "evidence-lookup"
+	baseKeyOutqueue = "evidence-outqueue"
+	baseKeyPending  = "evidence-pending"
+)
+
 func keyLookup(evidence types.Evidence) []byte {
-	return []byte(fmt.Sprintf("evidence-lookup/%d/%X", evidence.Height(), evidence.Hash()))
+	return []byte(fmt.Sprintf("%s/%d/%X", baseKeyLookup, evidence.Height(), evidence.Hash()))
 }
 
-func keyOutqueue(idx int, evidence types.Evidence) []byte {
-	return []byte(fmt.Sprintf("evidence-outqueue/%d/%d/%X", idx, evidence.Height(), evidence.Hash()))
+func keyOutqueue(evidence types.Evidence) []byte {
+	return []byte(fmt.Sprintf("%s/%d/%X", baseKeyOutqueue, evidence.Height(), evidence.Hash()))
 }
 
 func keyPending(evidence types.Evidence) []byte {
-	return []byte(fmt.Sprintf("evidence-pending/%d/%X", evidence.Height(), evidence.Hash()))
+	return []byte(fmt.Sprintf("%s/%d/%X", baseKeyPending, evidence.Height(), evidence.Hash()))
 }
 
 // EvidenceStore stores all the evidence we've seen, including
@@ -40,49 +46,85 @@ func keyPending(evidence types.Evidence) []byte {
 type EvidenceStore struct {
 	chainID string
 	db      dbm.DB
+
+	historicalValidators types.HistoricalValidators
 }
 
 func NewEvidenceStore(chainID string, db dbm.DB) *EvidenceStore {
 	return &EvidenceStore{
 		chainID: chainID,
 		db:      db,
+		// TODO historicalValidators
 	}
 }
 
+// PriorityEvidence returns the evidence from the outqueue, sorted by highest priority.
+func (store *EvidenceStore) PriorityEvidence() (evidence []types.Evidence) {
+	iter := store.db.IteratorPrefix([]byte(baseKeyOutqueue))
+	for iter.Next() {
+		val := iter.Value()
+
+		var ei evidenceInfo
+		wire.ReadBinaryBytes(val, &ei)
+		evidence = append(evidence, ei.Evidence)
+	}
+	// TODO: sort
+	return evidence
+}
+
+func (store *EvidenceStore) PendingEvidence() (evidence []types.Evidence) {
+	iter := store.db.IteratorPrefix([]byte(baseKeyPending))
+	for iter.Next() {
+		val := iter.Value()
+
+		var ei evidenceInfo
+		wire.ReadBinaryBytes(val, &ei)
+		evidence = append(evidence, ei.Evidence)
+	}
+	return evidence
+}
+
 // AddNewEvidence adds the given evidence to the database.
-func (store *EvidenceStore) AddNewEvidence(idx int, evidence types.Evidence) (bool, error) {
+func (store *EvidenceStore) AddNewEvidence(evidence types.Evidence) (bool, error) {
 	// check if we already have seen it
 	key := keyLookup(evidence)
 	v := store.db.Get(key)
-	if len(v) == 0 {
+	if len(v) != 0 {
 		return false, nil
 	}
 
-	// verify the evidence
-	if err := evidence.Verify(store.chainID); err != nil {
+	// verify evidence consistency
+	if err := evidence.Verify(store.chainID, store.historicalValidators); err != nil {
 		return false, err
 	}
 
-	// add it to the store
+	// TODO: or we let Verify return the val to avoid running this again?
+	valSet := store.historicalValidators.LoadValidators(evidence.Height())
+	_, val := valSet.GetByAddress(evidence.Address())
+	priority := int(val.VotingPower)
+
 	ei := evidenceInfo{
 		Committed: false,
-		Priority:  idx,
+		Priority:  priority,
 		Evidence:  evidence,
 	}
-	store.db.Set(key, wire.BinaryBytes(ei))
+	eiBytes := wire.BinaryBytes(ei)
 
-	key = keyOutqueue(idx, evidence)
-	store.db.Set(key, nullValue)
+	// add it to the store
+	store.db.Set(key, eiBytes)
+
+	key = keyOutqueue(evidence)
+	store.db.Set(key, eiBytes)
 
 	key = keyPending(evidence)
-	store.db.Set(key, nullValue)
+	store.db.Set(key, eiBytes)
 
 	return true, nil
 }
 
 // MarkEvidenceAsBroadcasted removes evidence from the outqueue.
-func (store *EvidenceStore) MarkEvidenceAsBroadcasted(idx int, evidence types.Evidence) {
-	key := keyOutqueue(idx, evidence)
+func (store *EvidenceStore) MarkEvidenceAsBroadcasted(evidence types.Evidence) {
+	key := keyOutqueue(evidence)
 	store.db.Delete(key)
 }