Merge pull request #7912 from multiversx/multi-proof-storage

multi proof pool

Author: AdoAdoAdo

common/constants.go

@@ -259,6 +259,10 @@ const MetricNumShardHeadersProcessed = "erd_num_shard_headers_processed"
 // MetricNumTimesInForkChoice is the metric that counts how many times a node was in fork choice
 const MetricNumTimesInForkChoice = "erd_fork_choice_count"
 
+// MetricNumEquivocationProofs is the metric that counts the equivocation events observed by the
+// proofs pool (different-hash proofs received for the same header nonce)
+const MetricNumEquivocationProofs = "erd_num_equivocation_proofs"
+
 // MetricHighestFinalBlock is the metric for the nonce of the highest final block
 const MetricHighestFinalBlock = "erd_highest_final_nonce"
 

dataRetriever/dataPool/proofsCache/export_test.go

@@ -2,6 +2,9 @@ package proofscache
 
 import "github.com/multiversx/mx-chain-core-go/data"
 
+// MaxProofsPerNonce -
+const MaxProofsPerNonce = maxProofsPerNonce
+
 // NewProofsCache -
 func NewProofsCache(bucketSize int) *proofsCache {
 	return newProofsCache(bucketSize)

dataRetriever/dataPool/proofsCache/proofsBucket.go

@@ -1,35 +1,50 @@
 package proofscache
 
-import "github.com/multiversx/mx-chain-core-go/data"
-
 type proofNonceBucket struct {
 	maxNonce      uint64
-	proofsByNonce map[uint64]string
+	proofsByNonce map[uint64][]string
 }
 
 func newProofBucket() *proofNonceBucket {
 	return &proofNonceBucket{
-		proofsByNonce: make(map[uint64]string),
+		proofsByNonce: make(map[uint64][]string),
 	}
 }
 
 func (p *proofNonceBucket) size() int {
-	return len(p.proofsByNonce)
+	size := 0
+	for _, hashes := range p.proofsByNonce {
+		size += len(hashes)
+	}
+
+	return size
 }
 
-func (p *proofNonceBucket) insert(proof data.HeaderProofHandler) string {
-	nonce := proof.GetHeaderNonce()
-	newHash := string(proof.GetHeaderHash())
+func (p *proofNonceBucket) hashesAt(nonce uint64) []string {
+	return p.proofsByNonce[nonce]
+}
 
-	oldHash, existed := p.proofsByNonce[nonce]
-	p.proofsByNonce[nonce] = newHash
+// insert adds the hash at the given nonce, keeping any different hashes already stored there
+func (p *proofNonceBucket) insert(nonce uint64, hash string) {
+	for _, existingHash := range p.proofsByNonce[nonce] {
+		if existingHash == hash {
+			return
+		}
+	}
+
+	p.proofsByNonce[nonce] = append(p.proofsByNonce[nonce], hash)
 
 	if nonce > p.maxNonce {
 		p.maxNonce = nonce
 	}
+}
 
-	if existed {
-		return oldHash
+func (p *proofNonceBucket) remove(nonce uint64, hash string) {
+	hashes := p.proofsByNonce[nonce]
+	for i, existingHash := range hashes {
+		if existingHash == hash {
+			p.proofsByNonce[nonce] = append(hashes[:i], hashes[i+1:]...)
+			return
+		}
 	}
-	return ""
 }

dataRetriever/dataPool/proofsCache/proofsCache.go

@@ -1,12 +1,18 @@
 package proofscache
 
 import (
+	"bytes"
+	"sort"
 	"sync"
 
 	"github.com/multiversx/mx-chain-core-go/core/check"
 	"github.com/multiversx/mx-chain-core-go/data"
 )
 
+// maxProofsPerNonce bounds how many different-hash proofs are kept at one nonce (more than one is
+// equivocation evidence); on overflow the highest (round, hash) proof is evicted
+const maxProofsPerNonce = 4
+
 type proofsCache struct {
 	mutProofsCache       sync.RWMutex
 	proofsByNonceBuckets map[uint64]*proofNonceBucket
@@ -34,54 +40,125 @@ func (pc *proofsCache) getProofByHash(headerHash []byte) (data.HeaderProofHandle
 	return proof, nil
 }
 
+// getProofByNonce returns the canonical proof at the given nonce: the lowest round, with the
+// lowest hash as tie-break
 func (pc *proofsCache) getProofByNonce(headerNonce uint64) (data.HeaderProofHandler, error) {
 	pc.mutProofsCache.RLock()
 	defer pc.mutProofsCache.RUnlock()
 
-	bucketKey := pc.getBucketKey(headerNonce)
-	bucket, ok := pc.proofsByNonceBuckets[bucketKey]
-	if !ok {
+	proof := pc.lowestProofAtNonce(headerNonce)
+	if proof == nil {
 		return nil, ErrMissingProof
 	}
 
-	proofHash, ok := bucket.proofsByNonce[headerNonce]
+	return proof, nil
+}
+
+// lowestProofAtNonce must be called under mutex protection; allocation-free on purpose, it sits
+// on the hot read path
+func (pc *proofsCache) lowestProofAtNonce(headerNonce uint64) data.HeaderProofHandler {
+	bucket, ok := pc.proofsByNonceBuckets[pc.getBucketKey(headerNonce)]
 	if !ok {
-		return nil, ErrMissingProof
+		return nil
+	}
+
+	var lowest data.HeaderProofHandler
+	for _, headerHash := range bucket.hashesAt(headerNonce) {
+		proof, hasProof := pc.proofsByHash[headerHash]
+		if !hasProof {
+			continue
+		}
+		if lowest == nil || lessProof(proof, lowest) {
+			lowest = proof
+		}
 	}
 
-	proof, ok := pc.proofsByHash[proofHash]
+	return lowest
+}
+
+// lessProof reports whether a orders before b by (round, hash) ascending
+func lessProof(a, b data.HeaderProofHandler) bool {
+	if a.GetHeaderRound() != b.GetHeaderRound() {
+		return a.GetHeaderRound() < b.GetHeaderRound()
+	}
+	return bytes.Compare(a.GetHeaderHash(), b.GetHeaderHash()) < 0
+}
+
+// getProofsByNonce returns all proofs at the given nonce, ordered by (round, hash) ascending
+func (pc *proofsCache) getProofsByNonce(headerNonce uint64) []data.HeaderProofHandler {
+	pc.mutProofsCache.RLock()
+	defer pc.mutProofsCache.RUnlock()
+
+	return pc.sortedProofsAtNonce(headerNonce)
+}
+
+// sortedProofsAtNonce must be called under mutex protection
+func (pc *proofsCache) sortedProofsAtNonce(headerNonce uint64) []data.HeaderProofHandler {
+	bucket, ok := pc.proofsByNonceBuckets[pc.getBucketKey(headerNonce)]
 	if !ok {
-		return nil, ErrMissingProof
+		return nil
 	}
 
-	return proof, nil
+	hashes := bucket.hashesAt(headerNonce)
+	proofs := make([]data.HeaderProofHandler, 0, len(hashes))
+	for _, headerHash := range hashes {
+		proof, hasProof := pc.proofsByHash[headerHash]
+		if hasProof {
+			proofs = append(proofs, proof)
+		}
+	}
+
+	if len(proofs) > 1 {
+		sort.Slice(proofs, func(i, j int) bool {
+			return lessProof(proofs[i], proofs[j])
+		})
+	}
+
+	return proofs
 }
 
-func (pc *proofsCache) addProof(proof data.HeaderProofHandler) {
+// addProof stores the proof, keeping different-hash proofs at the same nonce as equivocation
+// evidence; returns the competing proofs only on a newly stored hash, so equivocation reports once
+func (pc *proofsCache) addProof(proof data.HeaderProofHandler) []data.HeaderProofHandler {
 	if check.IfNil(proof) {
-		return
+		return nil
 	}
 
 	pc.mutProofsCache.Lock()
 	defer pc.mutProofsCache.Unlock()
 
-	oldHash := pc.insertProofByNonce(proof)
+	nonce := proof.GetHeaderNonce()
 	newHash := string(proof.GetHeaderHash())
+	bucket := pc.getOrCreateBucket(nonce)
+
+	alreadyStored := false
+	var competingProofs []data.HeaderProofHandler
+	for _, existingHash := range bucket.hashesAt(nonce) {
+		if existingHash == newHash {
+			alreadyStored = true
+			continue
+		}
 
-	// Delete the old hash from proofsByHash if it's different from the new hash
-	if len(oldHash) != 0 && oldHash != newHash {
-		log.Warn("proofsCache.addProof: overwrite proof by hash",
-			"oldHash", oldHash,
-			"newHash", newHash,
-		)
-		delete(pc.proofsByHash, oldHash)
+		existingProof, hasProof := pc.proofsByHash[existingHash]
+		if hasProof {
+			competingProofs = append(competingProofs, existingProof)
+		}
 	}
 
+	bucket.insert(nonce, newHash)
 	pc.proofsByHash[newHash] = proof
+
+	pc.evictExcessProofsAtNonce(bucket, nonce)
+
+	if alreadyStored {
+		return nil
+	}
+
+	return competingProofs
 }
 
 // addProofIfNoneAtNonce adds the proof only if its nonce slot is free; an occupied slot (same or
-// different hash) rejects the add and returns the pre-existing proof, never overwriting it
+// different hash) rejects the add and returns the pre-existing canonical proof, never overwriting it
 func (pc *proofsCache) addProofIfNoneAtNonce(proof data.HeaderProofHandler) (bool, data.HeaderProofHandler) {
 	if check.IfNil(proof) {
 		return false, nil
@@ -90,39 +167,57 @@ func (pc *proofsCache) addProofIfNoneAtNonce(proof data.HeaderProofHandler) (boo
 	pc.mutProofsCache.Lock()
 	defer pc.mutProofsCache.Unlock()
 
-	bucketKey := pc.getBucketKey(proof.GetHeaderNonce())
-	bucket, ok := pc.proofsByNonceBuckets[bucketKey]
-	if ok {
-		existingHash, hasNonce := bucket.proofsByNonce[proof.GetHeaderNonce()]
-		if hasNonce {
-			existingProof, hasProof := pc.proofsByHash[existingHash]
-			if hasProof {
-				return false, existingProof
-			}
-		}
+	nonce := proof.GetHeaderNonce()
+	existingProof := pc.lowestProofAtNonce(nonce)
+	if existingProof != nil {
+		return false, existingProof
 	}
 
-	_ = pc.insertProofByNonce(proof)
+	bucket := pc.getOrCreateBucket(nonce)
+	bucket.insert(nonce, string(proof.GetHeaderHash()))
 	pc.proofsByHash[string(proof.GetHeaderHash())] = proof
 
 	return true, nil
 }
 
+// evictExcessProofsAtNonce must be called under mutex protection
+func (pc *proofsCache) evictExcessProofsAtNonce(bucket *proofNonceBucket, nonce uint64) {
+	if len(bucket.hashesAt(nonce)) <= maxProofsPerNonce {
+		return
+	}
+
+	proofs := pc.sortedProofsAtNonce(nonce)
+	for len(proofs) > maxProofsPerNonce {
+		evictedProof := proofs[len(proofs)-1]
+		proofs = proofs[:len(proofs)-1]
+		evictedHash := string(evictedProof.GetHeaderHash())
+		bucket.remove(nonce, evictedHash)
+		delete(pc.proofsByHash, evictedHash)
+
+		log.Warn("proofsCache: too many proofs at the same nonce, evicted the highest round one",
+			"nonce", nonce,
+			"evicted hash", evictedProof.GetHeaderHash(),
+			"evicted round", evictedProof.GetHeaderRound(),
+		)
+	}
+}
+
 // getBucketKey will return bucket key as lower bound window value
 func (pc *proofsCache) getBucketKey(index uint64) uint64 {
 	return (index / pc.bucketSize) * pc.bucketSize
 }
 
-func (pc *proofsCache) insertProofByNonce(proof data.HeaderProofHandler) string {
-	bucketKey := pc.getBucketKey(proof.GetHeaderNonce())
+// getOrCreateBucket must be called under mutex protection
+func (pc *proofsCache) getOrCreateBucket(nonce uint64) *proofNonceBucket {
+	bucketKey := pc.getBucketKey(nonce)
 
 	bucket, ok := pc.proofsByNonceBuckets[bucketKey]
 	if !ok {
 		bucket = newProofBucket()
 		pc.proofsByNonceBuckets[bucketKey] = bucket
 	}
 
-	return bucket.insert(proof)
+	return bucket
 }
 
 func (pc *proofsCache) cleanupProofsBehindNonce(nonce uint64) {
@@ -142,7 +237,9 @@ func (pc *proofsCache) cleanupProofsBehindNonce(nonce uint64) {
 }
 
 func (pc *proofsCache) cleanupProofsInBucket(bucket *proofNonceBucket) {
-	for _, headerHash := range bucket.proofsByNonce {
-		delete(pc.proofsByHash, headerHash)
+	for _, headerHashes := range bucket.proofsByNonce {
+		for _, headerHash := range headerHashes {
+			delete(pc.proofsByHash, headerHash)
+		}
 	}
 }