Merge master into rc/supernova 2026.07.09

consensus/spos/bls/proxy/subroundsHandler_test.go

@@ -81,6 +81,7 @@ func getDefaultArgumentsSubroundHandler() (*SubroundsHandlerArgs, *spos.Consensu
 	consensusCore.SetScheduledProcessor(&consensus.ScheduledProcessorStub{})
 	consensusCore.SetMessageSigningHandler(&mock2.MessageSigningHandlerStub{})
 	consensusCore.SetPeerBlacklistHandler(&mock2.PeerBlacklistHandlerStub{})
+	consensusCore.SetPeerSignatureHandler(&cryptoMocks.PeerSignatureHandlerStub{})
 	consensusCore.SetSigningHandler(&consensus.SigningHandlerStub{})
 	consensusCore.SetEnableEpochsHandler(epochsEnable)
 	consensusCore.SetEnableRoundsHandler(&testscommon.EnableRoundsHandlerStub{})

consensus/spos/bls/v2/benchmark_send_proof_test.go

@@ -11,6 +11,7 @@ import (
 	"github.com/multiversx/mx-chain-crypto-go/signing/mcl"
 	mclMultisig "github.com/multiversx/mx-chain-crypto-go/signing/mcl/multisig"
 	"github.com/multiversx/mx-chain-crypto-go/signing/multisig"
+	"github.com/multiversx/mx-chain-go/dataRetriever/mock"
 	"github.com/stretchr/testify/require"
 
 	"github.com/multiversx/mx-chain-go/common"
@@ -166,6 +167,7 @@ func benchmarkSendProof(b *testing.B, numberOfKeys int) {
 				return consensusMetrics
 			},
 		},
+		&mock.ThrottlerStub{},
 	)
 	require.Nil(b, err)
 

consensus/spos/bls/v2/blsSubroundsFactory.go

@@ -294,6 +294,7 @@ func (fct *factory) generateEndRoundSubround() error {
 		fct.appStatusHandler,
 		fct.sentSignaturesTracker,
 		fct.worker,
+		fct.signatureThrottler,
 	)
 	if err != nil {
 		return err

consensus/spos/bls/v2/common.go

@@ -0,0 +1,27 @@
+package v2
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/multiversx/mx-chain-core-go/core"
+
+	"github.com/multiversx/mx-chain-go/consensus/spos"
+)
+
+func checkGoRoutinesThrottler(ctx context.Context, throttler core.Throttler) error {
+	for {
+		if throttler.CanProcess() {
+			break
+		}
+
+		select {
+		case <-time.After(time.Millisecond):
+			continue
+		case <-ctx.Done():
+			return fmt.Errorf("%w while checking the throttler", spos.ErrTimeIsOut)
+		}
+	}
+	return nil
+}

consensus/spos/bls/v2/common_test.go

@@ -0,0 +1,90 @@
+package v2
+
+import (
+	"context"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	dataRetrieverMock "github.com/multiversx/mx-chain-go/dataRetriever/mock"
+)
+
+func TestCheckGoRoutinesThrottler(t *testing.T) {
+	t.Parallel()
+
+	t.Run("throttler can process should return nil immediately", func(t *testing.T) {
+		t.Parallel()
+
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+
+		throttler := &dataRetrieverMock.ThrottlerStub{
+			CanProcessCalled: func() bool {
+				return true
+			},
+		}
+
+		err := checkGoRoutinesThrottler(ctx, throttler)
+		assert.Nil(t, err)
+	})
+
+	t.Run("throttler cannot process for a period of time, then can process, should return nil", func(t *testing.T) {
+		t.Parallel()
+
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+
+		var numCalls int32
+		throttler := &dataRetrieverMock.ThrottlerStub{
+			CanProcessCalled: func() bool {
+				return atomic.AddInt32(&numCalls, 1) >= 5
+			},
+		}
+
+		err := checkGoRoutinesThrottler(ctx, throttler)
+		assert.Nil(t, err)
+		assert.GreaterOrEqual(t, atomic.LoadInt32(&numCalls), int32(5))
+	})
+
+	t.Run("throttler cannot process should return error when context is done", func(t *testing.T) {
+		t.Parallel()
+
+		ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
+		defer cancel()
+
+		var numCalls int32
+		throttler := &dataRetrieverMock.ThrottlerStub{
+			CanProcessCalled: func() bool {
+				atomic.AddInt32(&numCalls, 1)
+				return false
+			},
+		}
+
+		err := checkGoRoutinesThrottler(ctx, throttler)
+		assert.NotNil(t, err)
+		assert.ErrorContains(t, err, "time is out")
+		assert.ErrorContains(t, err, "while checking the throttler")
+		assert.Greater(t, atomic.LoadInt32(&numCalls), int32(1))
+	})
+
+	t.Run("context already canceled and throttler cannot process should return error without retrying", func(t *testing.T) {
+		t.Parallel()
+
+		ctx, cancel := context.WithCancel(context.Background())
+		cancel()
+
+		var numCalls int32
+		throttler := &dataRetrieverMock.ThrottlerStub{
+			CanProcessCalled: func() bool {
+				atomic.AddInt32(&numCalls, 1)
+				return false
+			},
+		}
+
+		err := checkGoRoutinesThrottler(ctx, throttler)
+		assert.NotNil(t, err)
+		assert.ErrorContains(t, err, "time is out")
+	})
+}

consensus/spos/bls/v2/errors.go

@@ -17,6 +17,12 @@ var ErrValidSignatureFromInvalidSigner = errors.New("valid signature from invali
 // ErrHeaderHashMismatch signals that header hash does not match
 var ErrHeaderHashMismatch = errors.New("header hash does not match")
 
+// ErrPublicKeyMismatch signals that the BLS public key in the message does not belong to the transport sender
+var ErrPublicKeyMismatch = errors.New("public key does not match the message sender")
+
+// ErrSignerNotInConsensusGroup signals that the signer is not part of the current consensus group
+var ErrSignerNotInConsensusGroup = errors.New("signer is not part of the consensus group")
+
 // ErrNilRoundSyncController signals that a nil round sync controller has been provided
 var ErrNilRoundSyncController = errors.New("nil round sync controller")
 

consensus/spos/bls/v2/subroundEndRound.go

@@ -34,6 +34,7 @@
 	mutProcessingEndRound         sync.Mutex
 	sentSignatureTracker          spos.SentSignaturesTracker
 	worker                        spos.WorkerHandler
+	signatureThrottler            core.Throttler
 }
 
 // NewSubroundEndRound creates a subroundEndRound object
@@ -43,6 +44,7 @@
 	appStatusHandler core.AppStatusHandler,
 	sentSignatureTracker spos.SentSignaturesTracker,
 	worker spos.WorkerHandler,
+	signatureThrottler core.Throttler,
 ) (*subroundEndRound, error) {
 	err := checkNewSubroundEndRoundParams(baseSubround)
 	if err != nil {
@@ -57,6 +59,9 @@
 	if check.IfNil(worker) {
 		return nil, spos.ErrNilWorker
 	}
+	if check.IfNil(signatureThrottler) {
+		return nil, spos.ErrNilThrottler
+	}
 
 	srEndRound := subroundEndRound{
 		Subround:                      baseSubround,
@@ -65,6 +70,7 @@
 		mutProcessingEndRound:         sync.Mutex{},
 		sentSignatureTracker:          sentSignatureTracker,
 		worker:                        worker,
+		signatureThrottler:            signatureThrottler,
 	}
 	srEndRound.Job = srEndRound.doEndRoundJob
 	srEndRound.Check = srEndRound.doEndRoundConsensusCheck
@@ -224,11 +230,20 @@
 		return "", spos.ErrInvalidMessageType
 	}
 
+	if !sr.IsNodeInConsensusGroup(string(cnsMsg.PubKey)) {
+		return "", ErrSignerNotInConsensusGroup
+	}
+
 	err = sr.MessageSigningHandler().Verify(msg)
 	if err != nil {
 		return "", err
 	}
 
+	err = sr.PeerSignatureHandler().VerifyPeerSignature(cnsMsg.PubKey, msg.Peer(), cnsMsg.Signature)
+	if err != nil {
+		return "", ErrPublicKeyMismatch
+	}
+
 	if !bytes.Equal(headerHash, cnsMsg.BlockHeaderHash) {
 		return "", ErrHeaderHashMismatch
 	}
@@ -476,6 +491,10 @@
 	return bitmap, sig, nil
 }
 
+func (sr *subroundEndRound) checkGoRoutinesThrottler(ctx context.Context) error {
+	return checkGoRoutinesThrottler(ctx, sr.signatureThrottler)
+}
+
 // verifySignature implements parallel signature verification
 func (sr *subroundEndRound) verifySignature(i int, pk string, sigShare []byte) error {
 	err := sr.SigningHandler().VerifySignatureShare(uint16(i), sigShare, sr.GetData(), sr.GetHeader().GetEpoch())