fixes exec queue add and pop

process/asyncExecution/errors.go

@@ -10,3 +10,6 @@ var ErrNilExecutionTracker = errors.New("nil execution tracker")
 
 // ErrNilBlockProcessor signals that a nil block processor has been provided
 var ErrNilBlockProcessor = errors.New("nil block processor")
+
+// ErrNilExecutionResult signals that a nil execution result has been provided
+var ErrNilExecutionResult = errors.New("nil execution result")

process/asyncExecution/headersExecutor.go

@@ -17,6 +17,9 @@ var log = logger.GetOrCreate("process/asyncExecution")
 
 const timeToSleep = time.Millisecond * 5
 const timeToSleepOnError = time.Millisecond * 300
+const maxRetryAttempts = 10
+const maxBackoffTime = time.Second * 5
+const validationInterval = time.Minute * 1
 
 // ArgsHeadersExecutor holds all the components needed to create a new instance of *headersExecutor
 type ArgsHeadersExecutor struct {
@@ -72,6 +75,8 @@ func (he *headersExecutor) StartExecution() {
 
 // PauseExecution pauses the execution
 func (he *headersExecutor) PauseExecution() {
+	log.Debug("headersExecutor.PauseExecution: pausing execution")
+
 	he.mutPaused.Lock()
 	defer he.mutPaused.Unlock()
 
@@ -84,6 +89,8 @@ func (he *headersExecutor) PauseExecution() {
 
 // ResumeExecution resumes the execution
 func (he *headersExecutor) ResumeExecution() {
+	log.Debug("headersExecutor.ResumeExecution: resuming execution")
+
 	he.mutPaused.Lock()
 	defer he.mutPaused.Unlock()
 
@@ -93,10 +100,19 @@ func (he *headersExecutor) ResumeExecution() {
 func (he *headersExecutor) start(ctx context.Context) {
 	log.Debug("headersExecutor.start: starting execution")
 
+	validationTicker := time.NewTicker(validationInterval)
+	defer validationTicker.Stop()
+
 	for {
 		select {
 		case <-ctx.Done():
 			return
+		case <-validationTicker.C:
+			// Periodic queue validation
+			err := he.blocksQueue.ValidateQueueIntegrity()
+			if err != nil {
+				log.Error("headersExecutor.start: queue integrity validation failed", "err", err)
+			}
 		default:
 			he.mutPaused.RLock()
 			isPaused := he.isPaused
@@ -129,43 +145,74 @@ func (he *headersExecutor) start(ctx context.Context) {
 }
 
 func (he *headersExecutor) handleProcessError(ctx context.Context, pair queue.HeaderBodyPair) {
-	for {
+	retryCount := 0
+	backoffTime := timeToSleepOnError
+
+	for retryCount < maxRetryAttempts {
 		pairFromQueue, ok := he.blocksQueue.Peek()
 		if ok && pairFromQueue.Header.GetNonce() == pair.Header.GetNonce() {
 			// continue the processing (pop the next header from queue)
 			return
 		}
+
 		select {
 		case <-ctx.Done():
 			return
 		default:
 			// retry with the same pair
 			err := he.process(pair)
 			if err == nil {
+				log.Debug("headersExecutor.handleProcessError - retry succeeded",
+					"nonce", pair.Header.GetNonce(),
+					"retry_count", retryCount)
 				return
 			}
-			time.Sleep(timeToSleepOnError)
+			retryCount++
+			log.Warn("headersExecutor.handleProcessError - retry failed",
+				"nonce", pair.Header.GetNonce(),
+				"retry_count", retryCount,
+				"max_retries", maxRetryAttempts,
+				"err", err)
+
+			// Exponential backoff with maximum limit
+			time.Sleep(backoffTime)
+			backoffTime = backoffTime * 2
+			if backoffTime > maxBackoffTime {
+				backoffTime = maxBackoffTime
+			}
 		}
 	}
+
+	log.Error("headersExecutor.handleProcessError - max retries exceeded, skipping block",
+		"nonce", pair.Header.GetNonce(),
+		"max_retries", maxRetryAttempts)
 }
 
 func (he *headersExecutor) process(pair queue.HeaderBodyPair) error {
 	executionResult, err := he.blockProcessor.ProcessBlockProposal(pair.Header, pair.Body)
 	if err != nil {
 		log.Warn("headersExecutor.process process block failed",
 			"nonce", pair.Header.GetNonce(),
+			"prevHash", pair.Header.GetPrevHash(),
 			"err", err,
 		)
 		return err
 	}
 
+	// Validate execution result
+	if check.IfNil(executionResult) {
+		log.Warn("headersExecutor.process - nil execution result received",
+			"nonce", pair.Header.GetNonce())
+		return ErrNilExecutionResult
+	}
+
 	err = he.executionTracker.AddExecutionResult(executionResult)
 	if err != nil {
 		log.Warn("headersExecutor.process add execution result failed",
 			"nonce", pair.Header.GetNonce(),
 			"err", err,
 		)
-		return nil
+		return err
 	}
 
 	he.blockChain.SetFinalBlockInfo(

process/asyncExecution/headersExecutor_test.go

@@ -228,6 +228,11 @@ func TestHeadersExecutor_ProcessBlock(t *testing.T) {
 		blocksQueue := queue.NewBlocksQueue()
 		args.BlocksQueue = blocksQueue
 		wasAddExecutionResultCalled := atomicCore.Flag{}
+		args.BlockProcessor = &processMocks.BlockProcessorStub{
+			ProcessBlockProposalCalled: func(handler data.HeaderHandler, body data.BodyHandler) (data.BaseExecutionResultHandler, error) {
+				return &block.BaseExecutionResult{}, nil
+			},
+		}
 		args.ExecutionTracker = &processMocks.ExecutionTrackerStub{
 			AddExecutionResultCalled: func(executionResult data.BaseExecutionResultHandler) error {
 				wasAddExecutionResultCalled.SetValue(true)
@@ -393,7 +398,7 @@ func TestHeadersExecutor_Process(t *testing.T) {
 		require.Equal(t, expectedErr, err)
 	})
 
-	t.Run("should return nil on failing to add execution results to execution tracker", func(t *testing.T) {
+	t.Run("should return error on failing to add execution results to execution tracker", func(t *testing.T) {
 		t.Parallel()
 
 		args := createMockArgs()
@@ -422,7 +427,7 @@ func TestHeadersExecutor_Process(t *testing.T) {
 		}
 
 		err := executor.Process(pair)
-		require.Nil(t, err)
+		require.Equal(t, expectedErr, err)
 	})
 
 	t.Run("should add execution result info to blockchain handler", func(t *testing.T) {

process/asyncExecution/interface.go

@@ -10,6 +10,7 @@ import (
 type BlocksQueue interface {
 	Pop() (queue.HeaderBodyPair, bool)
 	Peek() (queue.HeaderBodyPair, bool)
+	ValidateQueueIntegrity() error
 	IsInterfaceNil() bool
 	Close()
 }

process/asyncExecution/queue/blocksQueue.go

@@ -6,8 +6,9 @@ import (
 
 	"github.com/multiversx/mx-chain-core-go/core/check"
 	"github.com/multiversx/mx-chain-core-go/data"
-	"github.com/multiversx/mx-chain-go/common"
 	logger "github.com/multiversx/mx-chain-logger-go"
+
+	"github.com/multiversx/mx-chain-go/common"
 )
 
 var log = logger.GetOrCreate("process/asyncExecution/queue")
@@ -70,13 +71,12 @@ func (bq *blocksQueue) AddOrReplace(pair HeaderBodyPair) error {
 
 	log.Debug("blocksQueue.AddOrReplace - block has been added", "nonce", pair.Header.GetNonce(), "queue size", len(bq.headerBodyPairs))
 
-	if len(bq.headerBodyPairs) > 1 {
-		return nil
-	}
-
-	select {
-	case bq.notifyCh <- struct{}{}:
-	default:
+	// Notify waiting Pop() calls when the first item is added to an empty queue
+	if len(bq.headerBodyPairs) == 1 {
+		select {
+		case bq.notifyCh <- struct{}{}:
+		default:
+		}
 	}
 
 	return nil
@@ -108,6 +108,7 @@ func (bq *blocksQueue) removeFromNonce(nonce uint64) []uint64 {
 		return removedNonces
 	}
 
+	log.Debug("blocksQueue.removeFromNonce - removing from nonce", "nonce", nonce, "num removed", len(pairsToBeRemoved))
 	for _, pair := range pairsToBeRemoved {
 		removedNonces = append(removedNonces, pair.Header.GetNonce())
 	}
@@ -138,10 +139,21 @@ func (bq *blocksQueue) add(pair HeaderBodyPair) error {
 // If the queue is empty, the method blocks until a new item is available.
 func (bq *blocksQueue) Pop() (HeaderBodyPair, bool) {
 	bq.mutex.Lock()
-	if len(bq.headerBodyPairs) > 1 {
+	if len(bq.headerBodyPairs) >= 1 {
 		item := bq.headerBodyPairs[0]
 		bq.headerBodyPairs = bq.headerBodyPairs[1:]
+		// Clear any stale notification from the channel when we take the fast path
+		if len(bq.headerBodyPairs) > 0 {
+			bq.mutex.Unlock()
+			return item, true
+		}
+
+		select {
+		case <-bq.notifyCh:
+		default:
+		}
 		bq.mutex.Unlock()
+
 		return item, true
 	}
 	if bq.closed {
@@ -190,6 +202,8 @@ func (bq *blocksQueue) Peek() (HeaderBodyPair, bool) {
 // RemoveAtNonceAndHigher removes the header-body pair at the specified nonce
 // and all pairs with higher nonces from the queue
 func (bq *blocksQueue) RemoveAtNonceAndHigher(nonce uint64) []uint64 {
+	log.Debug("blocksQueue.RemoveAtNonceAndHigher - removing from nonce and higher", "nonce", nonce)
+
 	bq.mutex.Lock()
 	defer bq.mutex.Unlock()
 
@@ -202,16 +216,54 @@ func (bq *blocksQueue) RemoveAtNonceAndHigher(nonce uint64) []uint64 {
 
 func (bq *blocksQueue) updateLastAddedNonceBasedOnRemovingNonce(removingNonce uint64) {
 	if len(bq.headerBodyPairs) > 0 {
-		bq.lastAddedNonce = bq.headerBodyPairs[len(bq.headerBodyPairs)-1].Header.GetNonce()
+		lastNonce := bq.headerBodyPairs[len(bq.headerBodyPairs)-1].Header.GetNonce()
+		bq.lastAddedNonce = lastNonce
+		log.Debug("blocksQueue.updateLastAddedNonceBasedOnRemovingNonce - updated from queue",
+			"new_last_nonce", lastNonce,
+			"queue_size", len(bq.headerBodyPairs))
 		return
 	}
 
 	if removingNonce > 0 {
 		bq.lastAddedNonce = removingNonce - 1
+		log.Debug("blocksQueue.updateLastAddedNonceBasedOnRemovingNonce - calculated from removing nonce",
+			"removing_nonce", removingNonce,
+			"new_last_nonce", bq.lastAddedNonce)
 		return
 	}
 
 	bq.lastAddedNonce = 0
+	log.Debug("blocksQueue.updateLastAddedNonceBasedOnRemovingNonce - reset to 0")
+}
+
+// ValidateQueueIntegrity checks the queue for consistency issues
+func (bq *blocksQueue) ValidateQueueIntegrity() error {
+	bq.mutex.Lock()
+	defer bq.mutex.Unlock()
+
+	if len(bq.headerBodyPairs) == 0 {
+		return nil
+	}
+
+	// Check that nonces are sequential
+	expectedNonce := bq.headerBodyPairs[0].Header.GetNonce()
+	for i, pair := range bq.headerBodyPairs {
+		currentNonce := pair.Header.GetNonce()
+		if currentNonce != expectedNonce {
+			return fmt.Errorf("%w: expected nonce %d at index %d, got %d", ErrQueueIntegrityViolation,
+				expectedNonce, i, currentNonce)
+		}
+		expectedNonce++
+	}
+
+	// Check that lastAddedNonce matches the last item
+	lastPair := bq.headerBodyPairs[len(bq.headerBodyPairs)-1]
+	if bq.lastAddedNonce != lastPair.Header.GetNonce() {
+		return fmt.Errorf("%w: lastAddedNonce %d doesn't match last pair nonce %d", ErrQueueIntegrityViolation,
+			bq.lastAddedNonce, lastPair.Header.GetNonce())
+	}
+
+	return nil
 }
 
 // Clean cleanup the queue and set the provided last added nonce
@@ -221,6 +273,8 @@ func (bq *blocksQueue) Clean(lastAddedNonce uint64) {
 
 	bq.headerBodyPairs = make([]HeaderBodyPair, 0)
 	bq.lastAddedNonce = lastAddedNonce
+
+	log.Debug("blocksQueue.Clean - queue cleaned", "last_added_nonce", lastAddedNonce)
 }
 
 // Close will close the queue