Fix seldom failing tests in hb v2

heartbeat/processor/export_test.go

@@ -0,0 +1,6 @@
+package processor
+
+// NewPeerAuthenticationRequestsProcessorWithoutGoRoutine -
+func NewPeerAuthenticationRequestsProcessorWithoutGoRoutine(args ArgPeerAuthenticationRequestsProcessor) (*peerAuthenticationRequestsProcessor, error) {
+	return newPeerAuthenticationRequestsProcessor(args)
+}

heartbeat/processor/peerAuthenticationRequestsProcessor.go

@@ -55,12 +55,26 @@ type peerAuthenticationRequestsProcessor struct {
 
 // NewPeerAuthenticationRequestsProcessor creates a new instance of peerAuthenticationRequestsProcessor
 func NewPeerAuthenticationRequestsProcessor(args ArgPeerAuthenticationRequestsProcessor) (*peerAuthenticationRequestsProcessor, error) {
+	processor, err := newPeerAuthenticationRequestsProcessor(args)
+	if err != nil {
+		return nil, err
+	}
+
+	var ctx context.Context
+	ctx, processor.cancel = context.WithTimeout(context.Background(), args.MaxTimeoutForRequests)
+
+	go processor.startRequestingMessages(ctx)
+
+	return processor, nil
+}
+
+func newPeerAuthenticationRequestsProcessor(args ArgPeerAuthenticationRequestsProcessor) (*peerAuthenticationRequestsProcessor, error) {
 	err := checkArgs(args)
 	if err != nil {
 		return nil, err
 	}
 
-	processor := &peerAuthenticationRequestsProcessor{
+	return &peerAuthenticationRequestsProcessor{
 		requestHandler:          args.RequestHandler,
 		nodesCoordinator:        args.NodesCoordinator,
 		peerAuthenticationPool:  args.PeerAuthenticationPool,
@@ -70,14 +84,7 @@ func NewPeerAuthenticationRequestsProcessor(args ArgPeerAuthenticationRequestsPr
 		delayBetweenRequests:    args.DelayBetweenRequests,
 		maxMissingKeysInRequest: args.MaxMissingKeysInRequest,
 		randomizer:              args.Randomizer,
-	}
-
-	var ctx context.Context
-	ctx, processor.cancel = context.WithTimeout(context.Background(), args.MaxTimeoutForRequests)
-
-	go processor.startRequestingMessages(ctx)
-
-	return processor, nil
+	}, nil
 }
 
 func checkArgs(args ArgPeerAuthenticationRequestsProcessor) error {

heartbeat/processor/peerAuthenticationRequestsProcessor_test.go

@@ -3,19 +3,22 @@ package processor
 import (
 	"bytes"
 	"errors"
+	"fmt"
 	"sort"
 	"strings"
 	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
 
+	mxAtomic "github.com/multiversx/mx-chain-core-go/core/atomic"
 	"github.com/multiversx/mx-chain-core-go/core/check"
 	"github.com/multiversx/mx-chain-core-go/core/random"
 	"github.com/multiversx/mx-chain-go/heartbeat"
 	"github.com/multiversx/mx-chain-go/testscommon"
 	"github.com/multiversx/mx-chain-go/testscommon/shardingMocks"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func createMockArgPeerAuthenticationRequestsProcessor() ArgPeerAuthenticationRequestsProcessor {
@@ -252,7 +255,7 @@ func TestPeerAuthenticationRequestsProcessor_isThresholdReached(t *testing.T) {
 		},
 	}
 
-	processor, err := NewPeerAuthenticationRequestsProcessor(args)
+	processor, err := NewPeerAuthenticationRequestsProcessorWithoutGoRoutine(args)
 	assert.Nil(t, err)
 	assert.False(t, check.IfNil(processor))
 
@@ -276,7 +279,7 @@ func TestPeerAuthenticationRequestsProcessor_requestMissingKeys(t *testing.T) {
 			},
 		}
 
-		processor, err := NewPeerAuthenticationRequestsProcessor(args)
+		processor, err := NewPeerAuthenticationRequestsProcessorWithoutGoRoutine(args)
 		assert.Nil(t, err)
 		assert.False(t, check.IfNil(processor))
 
@@ -293,7 +296,7 @@ func TestPeerAuthenticationRequestsProcessor_getRandMaxMissingKeys(t *testing.T)
 
 	args := createMockArgPeerAuthenticationRequestsProcessor()
 	args.MaxMissingKeysInRequest = 3
-	processor, err := NewPeerAuthenticationRequestsProcessor(args)
+	processor, err := NewPeerAuthenticationRequestsProcessorWithoutGoRoutine(args)
 	assert.Nil(t, err)
 	assert.False(t, check.IfNil(processor))
 
@@ -307,3 +310,58 @@ func TestPeerAuthenticationRequestsProcessor_getRandMaxMissingKeys(t *testing.T)
 		}
 	}
 }
+
+func TestPeerAuthenticationRequestsProcessor_goRoutineIsWorkingAndCloseShouldStopIt(t *testing.T) {
+	t.Parallel()
+
+	args := createMockArgPeerAuthenticationRequestsProcessor()
+	args.NodesCoordinator = &shardingMocks.NodesCoordinatorStub{
+		GetAllEligibleValidatorsPublicKeysCalled: func(epoch uint32) (map[uint32][][]byte, error) {
+			return map[uint32][][]byte{
+				0: {[]byte("pk0")},
+			}, nil
+		},
+	}
+	keysCalled := &mxAtomic.Flag{}
+	args.PeerAuthenticationPool = &testscommon.CacherStub{
+		KeysCalled: func() [][]byte {
+			keysCalled.SetValue(true)
+			return make([][]byte, 0)
+		},
+	}
+
+	processor, _ := NewPeerAuthenticationRequestsProcessor(args)
+	time.Sleep(args.DelayBetweenRequests*2 + time.Millisecond*300) // wait for the go routine to start and execute at least once
+	assert.True(t, keysCalled.IsSet())
+
+	err := processor.Close()
+	assert.Nil(t, err)
+
+	time.Sleep(time.Second) // wait for the go routine to stop
+	keysCalled.SetValue(false)
+
+	time.Sleep(args.DelayBetweenRequests*2 + time.Millisecond*300) // if the go routine did not stop it will set again the flag
+	assert.False(t, keysCalled.IsSet())
+}
+
+func TestPeerAuthenticationRequestsProcessor_CloseCalledTwiceShouldNotPanicNorError(t *testing.T) {
+	t.Parallel()
+
+	defer func() {
+		r := recover()
+		if r != nil {
+			require.Fail(t, fmt.Sprintf("should have not panicked: %v", r))
+		}
+	}()
+
+	args := createMockArgPeerAuthenticationRequestsProcessor()
+	processor, _ := NewPeerAuthenticationRequestsProcessor(args)
+
+	time.Sleep(args.DelayBetweenRequests*2 + time.Millisecond*300) // wait for the go routine to start and execute at least once
+
+	err := processor.Close()
+	assert.Nil(t, err)
+
+	err = processor.Close()
+	assert.Nil(t, err)
+}