Merge pull request #7357 from multiversx/revert-incoming-miniblocks

Revert incoming miniblocks

Author: AdoAdoAdo

process/block/gasConsumption.go

@@ -1,12 +1,14 @@
 package block
 
 import (
+	"bytes"
 	"fmt"
 	"sync"
 
 	"github.com/multiversx/mx-chain-core-go/core/check"
 	"github.com/multiversx/mx-chain-core-go/data"
 	"github.com/multiversx/mx-chain-go/process"
+	"golang.org/x/exp/slices"
 )
 
 // gasType defines the type of gas consumption
@@ -111,8 +113,12 @@ func (gc *gasConsumption) CheckIncomingMiniBlocks(
 	for i := 0; i < len(miniBlocks); i++ {
 		shouldSavePending, err = gc.checkIncomingMiniBlock(miniBlocks[i], transactions, bandwidthForIncomingMiniBlocks)
 		if shouldSavePending {
+			// saving pending starting with idx i, as it was not included either
 			gc.pendingMiniBlocks = append(gc.pendingMiniBlocks, miniBlocks[i:]...)
-			gc.transactionsForPendingMiniBlocks = transactions
+			for _, mb := range miniBlocks[i:] {
+				hashStr := string(mb.GetHash())
+				gc.transactionsForPendingMiniBlocks[hashStr] = transactions[hashStr]
+			}
 
 			return lastMiniBlockIndex, len(gc.pendingMiniBlocks), err
 		}
@@ -125,6 +131,48 @@ func (gc *gasConsumption) CheckIncomingMiniBlocks(
 	return lastMiniBlockIndex, 0, nil
 }
 
+// RevertIncomingMiniBlocks gets a list of mini block hashes and removes them from the local state
+func (gc *gasConsumption) RevertIncomingMiniBlocks(miniBlockHashes [][]byte) {
+	if len(miniBlockHashes) == 0 {
+		return
+	}
+
+	gc.mut.Lock()
+	defer gc.mut.Unlock()
+
+	for _, miniBlockHash := range miniBlockHashes {
+		// do not check here if it was found or not, as some pending mini blocks may be missing from this map
+		gasConsumedByMb := gc.gasConsumedByMiniBlock[string(miniBlockHash)]
+		delete(gc.gasConsumedByMiniBlock, string(miniBlockHash))
+
+		isPending, idxInPendingSlice := gc.isPendingMiniBlock(miniBlockHash)
+		if isPending {
+			gc.revertPendingMiniBlock(miniBlockHash, idxInPendingSlice)
+			continue
+		}
+
+		// if the mini block is not pending, remove it from the total gas consumed
+		gc.totalGasConsumed[incoming] -= gasConsumedByMb
+	}
+}
+
+func (gc *gasConsumption) isPendingMiniBlock(blockHash []byte) (bool, int) {
+	for idx, miniBlock := range gc.pendingMiniBlocks {
+		if bytes.Equal(miniBlock.GetHash(), blockHash) {
+			return true, idx
+		}
+	}
+
+	return false, initialLastIndex
+}
+
+func (gc *gasConsumption) revertPendingMiniBlock(miniBlockHash []byte, idxInPendingSlice int) {
+	// if the mini block was saved as pending, remove its transactions
+	// and remove it from pending slice
+	delete(gc.transactionsForPendingMiniBlocks, string(miniBlockHash))
+	gc.pendingMiniBlocks = slices.Delete(gc.pendingMiniBlocks, idxInPendingSlice, idxInPendingSlice+1)
+}
+
 func (gc *gasConsumption) checkIncomingMiniBlock(
 	mb data.MiniBlockHeaderHandler,
 	transactions map[string][]data.TransactionHandler,
@@ -153,13 +201,13 @@ func (gc *gasConsumption) checkIncomingMiniBlock(
 		return false, err
 	}
 
+	gc.gasConsumedByMiniBlock[string(mbHash)] = gasConsumedByMB
 	mbsLimitReached := gc.totalGasConsumed[incoming]+gasConsumedByMB > bandwidthForIncomingMiniBlocks
 	if !mbsLimitReached {
 		// limit not reached, continue
 		// this method might be called either from handling all mini blocks,
 		// either from handling pending, where the pending ones
 		// should have continuous indexes after the ones already included
-		gc.gasConsumedByMiniBlock[string(mbHash)] = gasConsumedByMB
 		gc.totalGasConsumed[incoming] += gasConsumedByMB
 
 		return false, nil

process/block/gasConsumption_test.go

@@ -630,13 +630,224 @@ func TestGasConsumption_ZeroIncomingLimit(t *testing.T) {
 	require.Zero(t, pendingMBs)      // added all
 }
 
+func TestGasConsumption_RevertIncomingMiniBlocks(t *testing.T) {
+	t.Parallel()
+
+	t.Run("empty mini block hashes should early exit", func(t *testing.T) {
+		t.Parallel()
+
+		gc, _ := block.NewGasConsumption(getMockArgsGasConsumption())
+		require.NotNil(t, gc)
+
+		gc.RevertIncomingMiniBlocks(nil)
+	})
+
+	t.Run("revert non-pending mini block should decrease total gas consumed", func(t *testing.T) {
+		t.Parallel()
+
+		gc, _ := block.NewGasConsumption(getMockArgsGasConsumption())
+		require.NotNil(t, gc)
+
+		// add mini blocks that will be within limits
+		mbs := generateMiniBlocks(3, 5)
+		txs := generateTxsForMiniBlocks(mbs)
+		lastMbIndex, pendingMbs, err := gc.CheckIncomingMiniBlocks(mbs, txs)
+		require.NoError(t, err)
+		require.Equal(t, 2, lastMbIndex)
+		require.Zero(t, pendingMbs)
+
+		totalGasBeforeRevert := gc.TotalGasConsumed()
+		expectedGasPerMb := maxGasLimitPerTx * 5 // 5 txs per mini block
+
+		// revert the first mini block and one none existing for coverage
+		gc.RevertIncomingMiniBlocks([][]byte{mbs[0].GetHash(), []byte("non-existent-hash")})
+
+		// total gas should decrease by the gas consumed by the first mini block
+		require.Equal(t, totalGasBeforeRevert-expectedGasPerMb, gc.TotalGasConsumed())
+
+		// revert the second and third mini blocks
+		gc.RevertIncomingMiniBlocks([][]byte{mbs[1].GetHash(), mbs[2].GetHash()})
+
+		// total gas should be zero now
+		require.Equal(t, uint64(0), gc.TotalGasConsumed())
+	})
+
+	t.Run("revert pending mini block at first position", func(t *testing.T) {
+		t.Parallel()
+
+		gc, _ := block.NewGasConsumption(getMockArgsGasConsumption())
+		require.NotNil(t, gc)
+
+		// add more mini blocks than the limit to create pending ones
+		mbs := generateMiniBlocks(10, 5)
+		txs := generateTxsForMiniBlocks(mbs)
+		lastMbIndex, pendingMbs, err := gc.CheckIncomingMiniBlocks(mbs, txs)
+		require.NoError(t, err)
+		require.Equal(t, 7, lastMbIndex)
+		require.Equal(t, 2, pendingMbs)
+
+		// get pending mini blocks
+		pendingMiniBlocks := gc.GetPendingMiniBlocks()
+		require.Len(t, pendingMiniBlocks, 2)
+
+		totalGasBeforeRevert := gc.TotalGasConsumed()
+
+		// revert the first pending mini block
+		firstPendingHash := mbs[8].GetHash()
+		gc.RevertIncomingMiniBlocks([][]byte{firstPendingHash})
+
+		// total gas should stay the same, only a pending mini block was removed
+		require.Equal(t, totalGasBeforeRevert, gc.TotalGasConsumed())
+
+		// check that pending mini blocks were updated
+		updatedPendingMiniBlocks := gc.GetPendingMiniBlocks()
+		require.Len(t, updatedPendingMiniBlocks, 1)
+		require.Equal(t, mbs[9].GetHash(), updatedPendingMiniBlocks[0].GetHash())
+	})
+
+	t.Run("revert pending mini block at last position", func(t *testing.T) {
+		t.Parallel()
+
+		gc, _ := block.NewGasConsumption(getMockArgsGasConsumption())
+		require.NotNil(t, gc)
+
+		// add more mini blocks than the limit to create pending ones
+		mbs := generateMiniBlocks(10, 5)
+		txs := generateTxsForMiniBlocks(mbs)
+		lastMbIndex, pendingMbs, err := gc.CheckIncomingMiniBlocks(mbs, txs)
+		require.NoError(t, err)
+		require.Equal(t, 7, lastMbIndex)
+		require.Equal(t, 2, pendingMbs)
+
+		// get pending mini blocks
+		pendingMiniBlocks := gc.GetPendingMiniBlocks()
+		require.Len(t, pendingMiniBlocks, 2)
+
+		totalGasBeforeRevert := gc.TotalGasConsumed()
+
+		// revert the last pending mini block (index 9, which is the last in pending)
+		lastPendingHash := mbs[9].GetHash()
+		gc.RevertIncomingMiniBlocks([][]byte{lastPendingHash})
+
+		// total gas should stay the same, only a pending mini block was removed
+		require.Equal(t, totalGasBeforeRevert, gc.TotalGasConsumed())
+
+		// check that pending mini blocks were updated
+		updatedPendingMiniBlocks := gc.GetPendingMiniBlocks()
+		require.Len(t, updatedPendingMiniBlocks, 1)
+		require.Equal(t, mbs[8].GetHash(), updatedPendingMiniBlocks[0].GetHash())
+	})
+
+	t.Run("revert pending mini block at middle position", func(t *testing.T) {
+		t.Parallel()
+
+		gc, _ := block.NewGasConsumption(getMockArgsGasConsumption())
+		require.NotNil(t, gc)
+
+		// add more mini blocks than the limit to create 3 pending ones
+		mbs := generateMiniBlocks(11, 5)
+		txs := generateTxsForMiniBlocks(mbs)
+		lastMbIndex, pendingMbs, err := gc.CheckIncomingMiniBlocks(mbs, txs)
+		require.NoError(t, err)
+		require.Equal(t, 7, lastMbIndex)
+		require.Equal(t, 3, pendingMbs)
+
+		// get pending mini blocks
+		pendingMiniBlocks := gc.GetPendingMiniBlocks()
+		require.Len(t, pendingMiniBlocks, 3)
+
+		totalGasBeforeRevert := gc.TotalGasConsumed()
+
+		// revert the middle pending mini block (index 9, which is the middle in pending)
+		middlePendingHash := mbs[9].GetHash()
+		gc.RevertIncomingMiniBlocks([][]byte{middlePendingHash})
+
+		// total gas should stay the same, only a pending mini block was removed
+		require.Equal(t, totalGasBeforeRevert, gc.TotalGasConsumed())
+
+		// check that pending mini blocks were updated
+		updatedPendingMiniBlocks := gc.GetPendingMiniBlocks()
+		require.Len(t, updatedPendingMiniBlocks, 2)
+		require.Equal(t, mbs[8].GetHash(), updatedPendingMiniBlocks[0].GetHash())
+		require.Equal(t, mbs[10].GetHash(), updatedPendingMiniBlocks[1].GetHash())
+	})
+
+	t.Run("revert multiple mini blocks with mixed pending and non-pending", func(t *testing.T) {
+		t.Parallel()
+
+		gc, _ := block.NewGasConsumption(getMockArgsGasConsumption())
+		require.NotNil(t, gc)
+
+		// add mini blocks, some will be pending
+		mbs := generateMiniBlocks(10, 5)
+		txs := generateTxsForMiniBlocks(mbs)
+		lastMbIndex, pendingMbs, err := gc.CheckIncomingMiniBlocks(mbs, txs)
+		require.NoError(t, err)
+		require.Equal(t, 7, lastMbIndex)
+		require.Equal(t, 2, pendingMbs)
+
+		totalGasBeforeRevert := gc.TotalGasConsumed()
+		expectedGasPerMb := maxGasLimitPerTx * 5
+
+		// revert both a non-pending mini block (index 0) and a pending one (index 8)
+		hashesToRevert := [][]byte{
+			mbs[0].GetHash(), // non-pending
+			mbs[8].GetHash(), // pending (first in pending)
+		}
+		gc.RevertIncomingMiniBlocks(hashesToRevert)
+
+		// total gas should decrease only by the non-pending mini block
+		require.Equal(t, totalGasBeforeRevert-expectedGasPerMb, gc.TotalGasConsumed())
+
+		// pending mini blocks should be reduced by one
+		updatedPendingMiniBlocks := gc.GetPendingMiniBlocks()
+		require.Len(t, updatedPendingMiniBlocks, 1)
+		require.Equal(t, mbs[9].GetHash(), updatedPendingMiniBlocks[0].GetHash())
+	})
+
+	t.Run("revert all pending mini blocks", func(t *testing.T) {
+		t.Parallel()
+
+		gc, _ := block.NewGasConsumption(getMockArgsGasConsumption())
+		require.NotNil(t, gc)
+
+		// add mini blocks with pending ones
+		mbs := generateMiniBlocks(10, 5)
+		txs := generateTxsForMiniBlocks(mbs)
+		lastMbIndex, pendingMbs, err := gc.CheckIncomingMiniBlocks(mbs, txs)
+		require.NoError(t, err)
+		require.Equal(t, 7, lastMbIndex)
+		require.Equal(t, 2, pendingMbs)
+
+		totalGasBeforeRevert := gc.TotalGasConsumed()
+
+		// revert all pending mini blocks
+		hashesToRevert := [][]byte{
+			mbs[8].GetHash(),
+			mbs[9].GetHash(),
+		}
+		gc.RevertIncomingMiniBlocks(hashesToRevert)
+
+		// total gas should stay the same, only pending mini blocks were removed
+		require.Equal(t, totalGasBeforeRevert, gc.TotalGasConsumed())
+
+		// no pending mini blocks should remain
+		updatedPendingMiniBlocks := gc.GetPendingMiniBlocks()
+		require.Len(t, updatedPendingMiniBlocks, 0)
+	})
+}
+
 func TestGasConsumption_ConcurrentOps(t *testing.T) {
 	if testing.Short() {
 		t.Skip("this is not a short test")
 	}
 
 	require.NotPanics(t, func() {
 		mbs := generateMiniBlocks(1, 2)
+		mbsHashes := make([][]byte, len(mbs))
+		for _, mb := range mbs {
+			mbsHashes = append(mbsHashes, mb.GetHash())
+		}
 		txsInMBs := generateTxsForMiniBlocks(mbs)
 
 		txHashes, txs := generateTxs(maxGasLimitPerTx, 3)
@@ -650,7 +861,7 @@ func TestGasConsumption_ConcurrentOps(t *testing.T) {
 
 		for i := 0; i < numCalls; i++ {
 			go func(idx int) {
-				switch idx % 11 {
+				switch idx % 12 {
 				case 0:
 					_, _, _ = gc.CheckOutgoingTransactions(txHashes, txs)
 				case 1:
@@ -673,6 +884,8 @@ func TestGasConsumption_ConcurrentOps(t *testing.T) {
 					gc.ZeroIncomingLimit()
 				case 10:
 					gc.ZeroOutgoingLimit()
+				case 11:
+					gc.RevertIncomingMiniBlocks(mbsHashes)
 				default:
 					require.Fail(t, "should have not been called")
 				}

process/interface.go

@@ -1537,6 +1537,7 @@ type GasComputation interface {
 		txHashes [][]byte,
 		transactions []data.TransactionHandler,
 	) (addedTxHashes [][]byte, pendingMiniBlocksAdded []data.MiniBlockHeaderHandler, err error)
+	RevertIncomingMiniBlocks(miniBlockHashes [][]byte)
 	GetBandwidthForTransactions() uint64
 	TotalGasConsumed() uint64
 	DecreaseIncomingLimit()

testscommon/gasComputationMock.go

@@ -14,6 +14,7 @@ type GasComputationMock struct {
 		txHashes [][]byte,
 		transactions []data.TransactionHandler,
 	) ([][]byte, []data.MiniBlockHeaderHandler, error)
+	RevertIncomingMiniBlocksCalled    func(miniBlockHashes [][]byte)
 	GetBandwidthForTransactionsCalled func() uint64
 	TotalGasConsumedCalled            func() uint64
 	DecreaseIncomingLimitCalled       func()
@@ -47,6 +48,13 @@ func (mock *GasComputationMock) CheckOutgoingTransactions(
 	return nil, nil, nil
 }
 
+// RevertIncomingMiniBlocks -
+func (mock *GasComputationMock) RevertIncomingMiniBlocks(miniBlockHashes [][]byte) {
+	if mock.RevertIncomingMiniBlocksCalled != nil {
+		mock.RevertIncomingMiniBlocksCalled(miniBlockHashes)
+	}
+}
+
 // GetBandwidthForTransactions -
 func (mock *GasComputationMock) GetBandwidthForTransactions() uint64 {
 	if mock.GetBandwidthForTransactionsCalled != nil {