Merge pull request #6501 from multiversx/integrate-proof-check-on-sync

Integrate proof check on sync

dataRetriever/dataPool/proofsCache/errors.go

@@ -7,3 +7,6 @@ var ErrMissingProof = errors.New("missing proof")
 
 // ErrNilProof signals that a nil proof has been provided
 var ErrNilProof = errors.New("nil proof provided")
+
+// ErrAlreadyExistingEquivalentProof signals that the provided proof was already exiting in the pool
+var ErrAlreadyExistingEquivalentProof = errors.New("already existing equivalent proof")

dataRetriever/dataPool/proofsCache/proofsPool.go

@@ -1,6 +1,7 @@
 package proofscache
 
 import (
+	"encoding/hex"
 	"fmt"
 	"sync"
 
@@ -14,12 +15,16 @@ var log = logger.GetOrCreate("dataRetriever/proofscache")
 type proofsPool struct {
 	mutCache sync.RWMutex
 	cache    map[uint32]*proofsCache
+
+	mutAddedProofSubscribers sync.RWMutex
+	addedProofSubscribers    []func(headerProof data.HeaderProofHandler)
 }
 
 // NewProofsPool creates a new proofs pool component
 func NewProofsPool() *proofsPool {
 	return &proofsPool{
-		cache: make(map[uint32]*proofsCache),
+		cache:                 make(map[uint32]*proofsCache),
+		addedProofSubscribers: make([]func(headerProof data.HeaderProofHandler), 0),
 	}
 }
 
@@ -36,8 +41,7 @@ func (pp *proofsPool) AddProof(
 
 	hasProof := pp.HasProof(shardID, headerHash)
 	if hasProof {
-		log.Trace("there was already a valid proof for header, headerHash: %s", headerHash)
-		return nil
+		return fmt.Errorf("%w, headerHash: %s", ErrAlreadyExistingEquivalentProof, hex.EncodeToString(headerHash))
 	}
 
 	pp.mutCache.Lock()
@@ -59,9 +63,20 @@ func (pp *proofsPool) AddProof(
 
 	proofsPerShard.addProof(headerProof)
 
+	pp.callAddedProofSubscribers(headerProof)
+
 	return nil
 }
 
+func (pp *proofsPool) callAddedProofSubscribers(headerProof data.HeaderProofHandler) {
+	pp.mutAddedProofSubscribers.RLock()
+	defer pp.mutAddedProofSubscribers.RUnlock()
+
+	for _, handler := range pp.addedProofSubscribers {
+		go handler(headerProof)
+	}
+}
+
 // CleanupProofsBehindNonce will cleanup proofs from pool based on nonce
 func (pp *proofsPool) CleanupProofsBehindNonce(shardID uint32, nonce uint64) error {
 	if nonce == 0 {
@@ -120,6 +135,18 @@ func (pp *proofsPool) HasProof(
 	return err == nil
 }
 
+// RegisterHandler registers a new handler to be called when a new data is added
+func (pp *proofsPool) RegisterHandler(handler func(headerProof data.HeaderProofHandler)) {
+	if handler == nil {
+		log.Error("attempt to register a nil handler to proofs pool")
+		return
+	}
+
+	pp.mutAddedProofSubscribers.Lock()
+	pp.addedProofSubscribers = append(pp.addedProofSubscribers, handler)
+	pp.mutAddedProofSubscribers.Unlock()
+}
+
 // IsInterfaceNil returns true if there is no value under the interface
 func (pp *proofsPool) IsInterfaceNil() bool {
 	return pp == nil

dataRetriever/dataPool/proofsCache/proofsPool_test.go

@@ -8,6 +8,7 @@ import (
 	"sync/atomic"
 	"testing"
 
+	"github.com/multiversx/mx-chain-core-go/data"
 	"github.com/multiversx/mx-chain-core-go/data/block"
 	proofscache "github.com/multiversx/mx-chain-go/dataRetriever/dataPool/proofsCache"
 	"github.com/stretchr/testify/assert"
@@ -65,6 +66,9 @@ func TestProofsPool_ShouldWork(t *testing.T) {
 	_ = pp.AddProof(proof3)
 	_ = pp.AddProof(proof4)
 
+	err := pp.AddProof(proof4)
+	require.True(t, errors.Is(err, proofscache.ErrAlreadyExistingEquivalentProof))
+
 	proof, err := pp.GetProof(shardID, []byte("hash3"))
 	require.Nil(t, err)
 	require.Equal(t, proof3, proof)
@@ -81,6 +85,28 @@ func TestProofsPool_ShouldWork(t *testing.T) {
 	require.Equal(t, proof4, proof)
 }
 
+func TestProofsPool_RegisterHandler(t *testing.T) {
+	t.Parallel()
+
+	pp := proofscache.NewProofsPool()
+
+	wasCalled := false
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+	handler := func(proof data.HeaderProofHandler) {
+		wasCalled = true
+		wg.Done()
+	}
+	pp.RegisterHandler(nil)
+	pp.RegisterHandler(handler)
+
+	_ = pp.AddProof(generateProof())
+
+	wg.Wait()
+
+	assert.True(t, wasCalled)
+}
+
 func TestProofsPool_Concurrency(t *testing.T) {
 	t.Parallel()
 
@@ -95,7 +121,7 @@ func TestProofsPool_Concurrency(t *testing.T) {
 
 	for i := 0; i < numOperations; i++ {
 		go func(idx int) {
-			switch idx % 5 {
+			switch idx % 6 {
 			case 0, 1, 2:
 				_ = pp.AddProof(generateProof())
 			case 3:
@@ -105,6 +131,10 @@ func TestProofsPool_Concurrency(t *testing.T) {
 				}
 			case 4:
 				_ = pp.CleanupProofsBehindNonce(generateRandomShardID(), generateRandomNonce())
+			case 5:
+				handler := func(proof data.HeaderProofHandler) {
+				}
+				pp.RegisterHandler(handler)
 			default:
 				assert.Fail(t, "should have not beed called")
 			}

epochStart/bootstrap/common.go

@@ -123,6 +123,9 @@ func checkArguments(args ArgsEpochStartBootstrap) error {
 	if check.IfNil(args.NodesCoordinatorRegistryFactory) {
 		return fmt.Errorf("%s: %w", baseErrorMessage, nodesCoordinator.ErrNilNodesCoordinatorRegistryFactory)
 	}
+	if check.IfNil(args.EnableEpochsHandler) {
+		return fmt.Errorf("%s: %w", baseErrorMessage, epochStart.ErrNilEnableEpochsHandler)
+	}
 
 	return nil
 }