Merge branch 'feat/supernova-async-exec' into supernova-bon

Author: raduchis

common/common.go

@@ -48,6 +48,11 @@ func PrepareOrderedTxHashesKey(headerHash []byte) []byte {
 	return append([]byte("execution"), headerHash...)
 }
 
+// PrepareUnexecutableTxHashesKey will prepare unexecutable transaction hashes key for cacher
+func PrepareUnexecutableTxHashesKey(headerHash []byte) []byte {
+	return append([]byte("unexecutable"), headerHash...)
+}
+
 // IsValidRelayedTxV3 returns true if the provided transaction is a valid transaction of type relayed v3
 func IsValidRelayedTxV3(tx data.TransactionHandler) bool {
 	relayedTx, isRelayedV3 := tx.(data.RelayedTransactionHandler)

common/commonCachedData.go

@@ -63,6 +63,23 @@ func GetCachedOrderedTxHashes(cache storage.Cacher, headerHash []byte) ([][]byte
 	return cachedDataSlice, nil
 }
 
+// GetCachedUnexecutableTxHashes will return the cached unexecutable tx hashes from the provided cache
+func GetCachedUnexecutableTxHashes(cache storage.Cacher, headerHash []byte) ([][]byte, error) {
+	unexecutableTxHashesKey := PrepareUnexecutableTxHashesKey(headerHash)
+	cachedData, ok := cache.Get(unexecutableTxHashesKey)
+	if !ok {
+		log.Warn("unexecutableTxHashes not found in dataPool", "hash", headerHash)
+		return nil, fmt.Errorf("%w for header %s", ErrMissingUnexecutableTxHash, hex.EncodeToString(headerHash))
+	}
+
+	cachedDataSlice, ok := cachedData.([][]byte)
+	if !ok {
+		return nil, fmt.Errorf("%w for cached unexecutable txs %s", ErrWrongTypeAssertion, hex.EncodeToString(headerHash))
+	}
+
+	return cachedDataSlice, nil
+}
+
 // GetCachedMbs will return the cached miniblocks from provided cache
 func GetCachedMbs(cache storage.Cacher, marshaller marshal.Marshalizer, headerHash []byte) ([]*block.MiniBlock, error) {
 	cachedIntraMBs, ok := cache.Get(headerHash)

common/commonCachedData_test.go

@@ -301,3 +301,41 @@ func TestGetCachedOrderedTxHashes(t *testing.T) {
 		require.Equal(t, hashes, res)
 	})
 }
+
+func TestGetCachedUnexecutableTxHashes(t *testing.T) {
+	t.Parallel()
+
+	t.Run("cannot find in cache should error", func(t *testing.T) {
+		t.Parallel()
+
+		cacher := cache.NewCacherMock()
+
+		headerHash := []byte("h")
+
+		_, err := GetCachedUnexecutableTxHashes(cacher, headerHash)
+		require.True(t, errors.Is(err, ErrMissingUnexecutableTxHash))
+	})
+
+	t.Run("wrong type in cache should error", func(t *testing.T) {
+		cacher := cache.NewCacherMock()
+
+		headerHash := []byte("h")
+		cacher.Put(PrepareUnexecutableTxHashesKey(headerHash), []byte("a"), 0)
+
+		_, err := GetCachedUnexecutableTxHashes(cacher, headerHash)
+		require.True(t, errors.Is(err, ErrWrongTypeAssertion))
+	})
+
+	t.Run("should work", func(t *testing.T) {
+		cacher := cache.NewCacherMock()
+
+		headerHash := []byte("h")
+		hashes := [][]byte{[]byte("a"), []byte("b"), []byte("c")}
+		cacher.Put(PrepareUnexecutableTxHashesKey(headerHash), hashes, 0)
+
+		res, err := GetCachedUnexecutableTxHashes(cacher, headerHash)
+		require.Nil(t, err)
+		require.Equal(t, hashes, res)
+	})
+
+}

common/common_test.go

@@ -1282,3 +1282,10 @@ func TestGetDeveloperFeesInEpoch(t *testing.T) {
 		require.Equal(t, big.NewInt(0), result)
 	})
 }
+
+func TestPrepareUnexecutableTxHashesKey(t *testing.T) {
+	t.Parallel()
+
+	hash := []byte("hash")
+	require.Equal(t, []byte("unexecutablehash"), common.PrepareUnexecutableTxHashesKey(hash))
+}

common/errors.go

@@ -67,3 +67,6 @@ var ErrMissingOrderedTxHashes = errors.New("missing ordered tx hashes")
 
 // ErrInvalidHeader signals that an invalid header has been provided
 var ErrInvalidHeader = errors.New("invalid header")
+
+// ErrMissingUnexecutableTxHash signals that unexecutable tx hashes are missing
+var ErrMissingUnexecutableTxHash = errors.New("missing unexecutable tx hash")

epochStart/metachain/rewardsV2.go

@@ -161,7 +161,7 @@ func (rc *rewardsCreatorV2) createRewardsMiniBlocks(
 	rc.clean()
 	rc.flagDelegationSystemSCEnabled.SetValue(args.newEpoch >= rc.enableEpochsHandler.GetActivationEpoch(common.StakingV2Flag))
 
-	protRwdTx, protRwdShardId, err := rc.createProtocolSustainabilityRewardTransaction(args.newEpoch, args.round, args.computedEconomics.GetRewardsForProtocolSustainability())
+	protRwdTx, protRwdShardId, err := rc.createProtocolSustainabilityRewardTransaction(args.newEpoch, args.round, rc.economicsDataProvider.RewardsForProtocolSustainability())
 	if err != nil {
 		return nil, err
 	}

factory/disabled/txCoordinator.go

@@ -20,6 +20,11 @@ func (txCoordinator *TxCoordinator) CreatePostProcessMiniBlocks() block.MiniBloc
 	return make(block.MiniBlockSlice, 0)
 }
 
+// GetUnExecutableTransactions does nothing as it is disabled
+func (txCoordinator *TxCoordinator) GetUnExecutableTransactions() map[string]struct{} {
+	return make(map[string]struct{})
+}
+
 // CreateReceiptsHash does nothing as it is disabled
 func (txCoordinator *TxCoordinator) CreateReceiptsHash() ([]byte, error) {
 	return nil, nil

integrationTests/mock/transactionCoordinatorMock.go

@@ -40,6 +40,7 @@ type TransactionCoordinatorMock struct {
 	AddTxsFromMiniBlocksCalled                           func(miniBlocks block.MiniBlockSlice)
 	AddTransactionsCalled                                func(txHandlers []data.TransactionHandler, blockType block.Type)
 	ComputeTransactionTypeInEpochCalled                  func(tx data.TransactionHandler, epoch uint32) (process.TransactionType, process.TransactionType, bool)
+	GetUnExecutableTransactionsCalled                    func() map[string]struct{}
 }
 
 // GetAllCurrentLogs -
@@ -114,6 +115,15 @@ func (tcm *TransactionCoordinatorMock) SaveTxsToStorage(body *block.Body) {
 	tcm.SaveTxsToStorageCalled(body)
 }
 
+// GetUnExecutableTransactions -
+func (tcm *TransactionCoordinatorMock) GetUnExecutableTransactions() map[string]struct{} {
+	if tcm.GetUnExecutableTransactionsCalled != nil {
+		return tcm.GetUnExecutableTransactionsCalled()
+	}
+
+	return nil
+}
+
 // RestoreBlockDataFromStorage -
 func (tcm *TransactionCoordinatorMock) RestoreBlockDataFromStorage(body *block.Body) (int, error) {
 	if tcm.RestoreBlockDataFromStorageCalled == nil {

node/chainSimulator/process/processor.go

@@ -7,12 +7,12 @@ import (
 	"github.com/multiversx/mx-chain-core-go/core/check"
 	"github.com/multiversx/mx-chain-core-go/data"
 	dataBlock "github.com/multiversx/mx-chain-core-go/data/block"
-	"github.com/multiversx/mx-chain-go/node/chainSimulator/dtos"
 	logger "github.com/multiversx/mx-chain-logger-go"
 
 	"github.com/multiversx/mx-chain-go/common"
 	heartbeatData "github.com/multiversx/mx-chain-go/heartbeat/data"
 	"github.com/multiversx/mx-chain-go/node/chainSimulator/configs"
+	"github.com/multiversx/mx-chain-go/node/chainSimulator/dtos"
 	"github.com/multiversx/mx-chain-go/process/asyncExecution/queue"
 	"github.com/multiversx/mx-chain-go/sharding/nodesCoordinator"
 )
@@ -127,6 +127,11 @@ func (creator *blocksCreator) CreateNewBlock() (*dtos.BroadcastData, error) {
 		return nil, err
 	}
 
+	log.Debug("Leader in current block",
+		"shardID", newHeader.GetShardID(),
+		"round", newHeader.GetRound(),
+		"leader", leader.PubKey())
+
 	pubKeyBitmap := GeneratePubKeyBitmap(len(validators))
 	for idx, validator := range validators {
 		isManaged := cryptoComponents.KeysHandler().IsKeyManagedByCurrentNode(validator.PubKey())

node/external/transactionAPI/apiTransactionProcessor.go

@@ -1,6 +1,7 @@
 package transactionAPI
 
 import (
+	"bytes"
 	"encoding/hex"
 	"errors"
 	"fmt"
@@ -748,20 +749,53 @@ func (atp *apiTransactionProcessor) computeTimestampForRoundAsMs(round uint64) i
 	return timestamp.UnixMilli()
 }
 
-func (atp *apiTransactionProcessor) checkExecutionResult(miniblockMetadata *dblookupext.MiniblockMetadata) error {
+func (atp *apiTransactionProcessor) checkExecutionResultAndTx(miniblockMetadata *dblookupext.MiniblockMetadata) (bool, error) {
 	isSupernovaEnabled := atp.enableRoundsHandler.IsFlagEnabledInRound(common.SupernovaRoundFlag, miniblockMetadata.Round)
 	if !isSupernovaEnabled {
-		return nil
+		return true, nil
 	}
 
 	headerHash := miniblockMetadata.GetHeaderHash()
 	executionResultsStorer, errG := atp.storageService.GetStorer(dataRetriever.ExecutionResultsUnit)
 	if errG != nil {
-		return errG
+		return false, errG
+	}
+
+	executionResultsBytes, err := executionResultsStorer.GetFromEpoch(headerHash, miniblockMetadata.GetEpoch())
+	if err != nil {
+		return false, err
+	}
+
+	if atp.shardCoordinator.SelfId() == core.MetachainShardId {
+		// we cannot have unexecutable txs on metachain
+		return true, nil
+	}
+	mbHeaders, err := atp.getMbHeadersFromExecutionResultBytes(executionResultsBytes)
+	if err != nil {
+		return false, err
 	}
 
-	_, err := executionResultsStorer.GetFromEpoch(headerHash, miniblockMetadata.GetEpoch())
-	return err
+	// check if the transaction miniblock metadata has a mb header on execution result
+	// if yes - the transaction was executed
+	// if no  - the transaction was proposed but not executed
+	currentTxIsExecuted := false
+	for _, mbHeader := range mbHeaders {
+		if bytes.Equal(mbHeader.Hash, miniblockMetadata.MiniblockHash) {
+			currentTxIsExecuted = true
+			break
+		}
+	}
+	return currentTxIsExecuted, nil
+}
+
+func (atp *apiTransactionProcessor) getMbHeadersFromExecutionResultBytes(executionResultBytes []byte) ([]block.MiniBlockHeader, error) {
+	executResult := &block.ExecutionResult{}
+	err := atp.marshalizer.Unmarshal(executResult, executionResultBytes)
+	if err != nil {
+		return nil, err
+	}
+
+	return executResult.GetMiniBlockHeaders(), nil
 }
 
 func (atp *apiTransactionProcessor) lookupHistoricalTransaction(hash []byte, withResults bool) (*transaction.ApiTransactionResult, error) {
@@ -770,7 +804,7 @@ func (atp *apiTransactionProcessor) lookupHistoricalTransaction(hash []byte, wit
 		return nil, fmt.Errorf("%s: %w", ErrTransactionNotFound.Error(), err)
 	}
 
-	err = atp.checkExecutionResult(miniblockMetadata)
+	isExecuted, err := atp.checkExecutionResultAndTx(miniblockMetadata)
 	if err != nil {
 		return nil, fmt.Errorf("%s: %w", ErrTransactionNotFound.Error(), err)
 	}
@@ -802,6 +836,11 @@ func (atp *apiTransactionProcessor) lookupHistoricalTransaction(hash []byte, wit
 		return nil, fmt.Errorf("%s: %w", ErrNilStatusComputer.Error(), err)
 	}
 
+	if !isExecuted {
+		tx.Status = transaction.TxStatusNotExecutable
+		return tx, nil
+	}
+
 	if ok, _ := statusComputer.SetStatusIfIsRewardReverted(
 		tx,
 		block.Type(miniblockMetadata.Type),