Validate Concurrent Messages + Update BankSend

aclmapping/bank/mapping_test.go

@@ -0,0 +1,155 @@
+package aclbankmapping
+
+import (
+	"testing"
+
+	"github.com/cosmos/cosmos-sdk/crypto/keys/secp256k1"
+	"github.com/cosmos/cosmos-sdk/simapp"
+	sdk "github.com/cosmos/cosmos-sdk/types"
+	sdkacltypes "github.com/cosmos/cosmos-sdk/types/accesscontrol"
+	acltypes "github.com/cosmos/cosmos-sdk/x/accesscontrol/types"
+	authtypes "github.com/cosmos/cosmos-sdk/x/auth/types"
+	"github.com/cosmos/cosmos-sdk/x/bank"
+	"github.com/cosmos/cosmos-sdk/x/bank/types"
+	banktypes "github.com/cosmos/cosmos-sdk/x/bank/types"
+	oracletypes "github.com/sei-protocol/sei-chain/x/oracle/types"
+	"github.com/stretchr/testify/require"
+	tmproto "github.com/tendermint/tendermint/proto/tendermint/types"
+)
+
+
+func cacheTxContext(ctx sdk.Context) (sdk.Context, sdk.CacheMultiStore) {
+	ms := ctx.MultiStore()
+	// TODO: https://github.com/cosmos/cosmos-sdk/issues/2824
+	msCache := ms.CacheMultiStore()
+	return ctx.WithMultiStore(msCache), msCache
+}
+
+func TestMsgBankSendAclOps(t *testing.T) {
+	priv1 := secp256k1.GenPrivKey()
+	addr1 := sdk.AccAddress(priv1.PubKey().Address())
+	priv2 := secp256k1.GenPrivKey()
+	addr2 := sdk.AccAddress(priv2.PubKey().Address())
+	coins     := sdk.Coins{sdk.NewInt64Coin("foocoin", 10)}
+
+	tests := []struct {
+		name          string
+		expectedError error
+		msg           *types.MsgSend
+		dynamicDep 	  bool
+	}{
+		{
+			name:          "default send",
+			msg:           types.NewMsgSend(addr1, addr2, coins),
+			expectedError: nil,
+			dynamicDep: true,
+		},
+		{
+			name:          "dont check synchronous",
+			msg:           types.NewMsgSend(addr1, addr2, coins),
+			expectedError: nil,
+			dynamicDep: false,
+		},
+	}
+
+	acc1 := &authtypes.BaseAccount{
+		Address: addr1.String(),
+	}
+	acc2 := &authtypes.BaseAccount{
+		Address: addr2.String(),
+	}
+	accs := authtypes.GenesisAccounts{acc1, acc2}
+	balances := []types.Balance{
+		{
+			Address: addr1.String(),
+			Coins:   coins,
+		},
+		{
+			Address: addr2.String(),
+			Coins:   coins,
+		},
+	}
+
+	app := simapp.SetupWithGenesisAccounts(accs, balances...)
+	ctx := app.BaseApp.NewContext(false, tmproto.Header{})
+
+	handler := bank.NewHandler(app.BankKeeper)
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			handlerCtx, cms := cacheTxContext(ctx)
+			_, err := handler(handlerCtx, tc.msg)
+
+			depdenencies , _ := MsgSendDependencyGenerator(app.AccessControlKeeper, handlerCtx, tc.msg)
+
+			if !tc.dynamicDep {
+				depdenencies = sdkacltypes.GetDefaultSynchronousAccessOps()
+			}
+
+			if tc.expectedError != nil {
+				require.EqualError(t, err, tc.expectedError.Error())
+			} else {
+				require.NoError(t, err)
+			}
+			missing := (sdkacltypes.ValidateAccessOperations(depdenencies, cms.GetEvents()))
+			require.Empty(t, missing)
+		})
+	}
+}
+
+func TestGeneratorInvalidMessageTypes(t *testing.T) {
+	accs := authtypes.GenesisAccounts{}
+	balances := []types.Balance{}
+
+	app := simapp.SetupWithGenesisAccounts(accs, balances...)
+	ctx := app.BaseApp.NewContext(false, tmproto.Header{})
+
+	oracleVote := oracletypes.MsgAggregateExchangeRateVote{
+		ExchangeRates: "1usei",
+		Feeder:        "test",
+		Validator:     "validator",
+	}
+
+	_, err := MsgSendDependencyGenerator(app.AccessControlKeeper, ctx, &oracleVote)
+	require.Error(t, err)
+}
+
+func TestMsgBeginBankSendGenerator(t *testing.T) {
+	priv1 := secp256k1.GenPrivKey()
+	addr1 := sdk.AccAddress(priv1.PubKey().Address())
+	priv2 := secp256k1.GenPrivKey()
+	addr2 := sdk.AccAddress(priv2.PubKey().Address())
+	coins     := sdk.Coins{sdk.NewInt64Coin("foocoin", 10)}
+
+	acc1 := &authtypes.BaseAccount{
+		Address: addr1.String(),
+	}
+	acc2 := &authtypes.BaseAccount{
+		Address: addr2.String(),
+	}
+	accs := authtypes.GenesisAccounts{acc1, acc2}
+	balances := []types.Balance{
+		{
+			Address: addr1.String(),
+			Coins:   coins,
+		},
+		{
+			Address: addr2.String(),
+			Coins:   coins,
+		},
+	}
+
+	app := simapp.SetupWithGenesisAccounts(accs, balances...)
+	ctx := app.BaseApp.NewContext(false, tmproto.Header{})
+
+	sendMsg := banktypes.MsgSend{
+		FromAddress: addr1.String(),
+		ToAddress: addr2.String(),
+		Amount: coins,
+	}
+
+	accessOps, err := MsgSendDependencyGenerator(app.AccessControlKeeper, ctx, &sendMsg)
+	require.NoError(t, err)
+	err = acltypes.ValidateAccessOps(accessOps)
+	require.NoError(t, err)
+}

aclmapping/bank/mappings.go

@@ -7,6 +7,7 @@ import (
 	sdkacltypes "github.com/cosmos/cosmos-sdk/types/accesscontrol"
 	aclkeeper "github.com/cosmos/cosmos-sdk/x/accesscontrol/keeper"
 	acltypes "github.com/cosmos/cosmos-sdk/x/accesscontrol/types"
+	authtypes "github.com/cosmos/cosmos-sdk/x/auth/types"
 	banktypes "github.com/cosmos/cosmos-sdk/x/bank/types"
 	utils "github.com/sei-protocol/sei-chain/aclmapping/utils"
 )
@@ -23,12 +24,13 @@ func GetBankDepedencyGenerator() aclkeeper.DependencyGeneratorMap {
 	return dependencyGeneratorMap
 }
 
-// TODO:: we can make resource types more granular  (e.g KV_PARAM or KV_BANK_BALANCE)
 func MsgSendDependencyGenerator(keeper aclkeeper.Keeper, ctx sdk.Context, msg sdk.Msg) ([]sdkacltypes.AccessOperation, error) {
 	msgSend, ok := msg.(*banktypes.MsgSend)
 	if !ok {
 		return []sdkacltypes.AccessOperation{}, ErrorInvalidMsgType
 	}
+	fromAddrIdentifier := string(banktypes.CreateAccountBalancesPrefixFromBech32(msgSend.FromAddress))
+	toAddrIdentifier := string(banktypes.CreateAccountBalancesPrefixFromBech32(msgSend.ToAddress))
 
 	accessOperations := []sdkacltypes.AccessOperation{
 		// MsgSend also checks if the coin denom is enabled, but the information is from the params.
@@ -37,46 +39,53 @@ func MsgSendDependencyGenerator(keeper aclkeeper.Keeper, ctx sdk.Context, msg sd
 		// Checks balance of sender
 		{
 			AccessType:         sdkacltypes.AccessType_READ,
-			ResourceType:       sdkacltypes.ResourceType_KV,
-			IdentifierTemplate: utils.GetIdentifierTemplatePerModule(utils.BANK, msgSend.FromAddress),
+			ResourceType:       sdkacltypes.ResourceType_KV_BANK,
+			IdentifierTemplate: fromAddrIdentifier,
 		},
 		// Reduce the amount from the sender's balance
 		{
 			AccessType:         sdkacltypes.AccessType_WRITE,
-			ResourceType:       sdkacltypes.ResourceType_KV,
-			IdentifierTemplate: utils.GetIdentifierTemplatePerModule(utils.BANK, msgSend.FromAddress),
+			ResourceType:       sdkacltypes.ResourceType_KV_BANK,
+			IdentifierTemplate: fromAddrIdentifier,
 		},
 
 		// Checks balance for receiver
 		{
 			AccessType:         sdkacltypes.AccessType_READ,
-			ResourceType:       sdkacltypes.ResourceType_KV,
-			IdentifierTemplate: utils.GetIdentifierTemplatePerModule(utils.BANK, msgSend.ToAddress),
+			ResourceType:       sdkacltypes.ResourceType_KV_BANK,
+			IdentifierTemplate: toAddrIdentifier,
 		},
 		{
 			AccessType:         sdkacltypes.AccessType_WRITE,
-			ResourceType:       sdkacltypes.ResourceType_KV,
-			IdentifierTemplate: utils.GetIdentifierTemplatePerModule(utils.BANK, msgSend.ToAddress),
+			ResourceType:       sdkacltypes.ResourceType_KV_BANK,
+			IdentifierTemplate: toAddrIdentifier,
 		},
 
 		// Tries to create the reciever's account if it doesn't exist
 		{
 			AccessType:         sdkacltypes.AccessType_READ,
-			ResourceType:       sdkacltypes.ResourceType_KV,
-			IdentifierTemplate: utils.GetIdentifierTemplatePerModule(utils.AUTH, msgSend.ToAddress),
+			ResourceType:       sdkacltypes.ResourceType_KV_AUTH,
+			IdentifierTemplate: string(authtypes.CreateAddressStoreKeyFromBech32(msgSend.ToAddress)),
 		},
 		{
 			AccessType:         sdkacltypes.AccessType_WRITE,
-			ResourceType:       sdkacltypes.ResourceType_KV,
-			IdentifierTemplate: utils.GetIdentifierTemplatePerModule(utils.AUTH, msgSend.ToAddress),
+			ResourceType:       sdkacltypes.ResourceType_KV_AUTH,
+			IdentifierTemplate: string(authtypes.CreateAddressStoreKeyFromBech32(msgSend.ToAddress)),
+		},
+
+		// Gets Account Info for the sender
+		{
+			AccessType:         sdkacltypes.AccessType_READ,
+			ResourceType:       sdkacltypes.ResourceType_KV_AUTH,
+			IdentifierTemplate: string(authtypes.CreateAddressStoreKeyFromBech32(msgSend.FromAddress)),
 		},
 
-		// Last Operation should always be a commit
 		{
 			ResourceType:       sdkacltypes.ResourceType_ANY,
 			AccessType:         sdkacltypes.AccessType_COMMIT,
 			IdentifierTemplate: utils.DefaultIDTemplate,
 		},
 	}
+
 	return accessOperations, nil
 }

aclmapping/utils/identifier_templates.go

@@ -1,4 +1,4 @@
-package util
+package utils
 
 import (
 	"fmt"

app/app.go

@@ -32,15 +32,15 @@ import (
 	"github.com/cosmos/cosmos-sdk/simapp"
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	sdkacltypes "github.com/cosmos/cosmos-sdk/types/accesscontrol"
+	sdkerrors "github.com/cosmos/cosmos-sdk/types/errors"
 	"github.com/cosmos/cosmos-sdk/types/module"
 	"github.com/cosmos/cosmos-sdk/version"
-	"github.com/cosmos/cosmos-sdk/x/auth"
-	"github.com/cosmos/cosmos-sdk/x/auth/ante"
-
 	aclmodule "github.com/cosmos/cosmos-sdk/x/accesscontrol"
 	aclclient "github.com/cosmos/cosmos-sdk/x/accesscontrol/client"
 	aclkeeper "github.com/cosmos/cosmos-sdk/x/accesscontrol/keeper"
 	acltypes "github.com/cosmos/cosmos-sdk/x/accesscontrol/types"
+	"github.com/cosmos/cosmos-sdk/x/auth"
+	"github.com/cosmos/cosmos-sdk/x/auth/ante"
 
 	authrest "github.com/cosmos/cosmos-sdk/x/auth/client/rest"
 	authkeeper "github.com/cosmos/cosmos-sdk/x/auth/keeper"
@@ -996,24 +996,29 @@ func (app *App) ProcessTxConcurrent(
 	resultChan chan<- ChannelResult,
 	txCompletionSignalingMap acltypes.MessageCompletionSignalMapping,
 	txBlockingSignalsMap acltypes.MessageCompletionSignalMapping,
+	// txMsgAccessOpMapping acltypes.MsgIndexToAccessOpMapping,
 ) {
 	defer wg.Done()
 	// Store the Channels in the Context Object for each transaction
-	ctx = ctx.WithTxBlockingChannels(getChannelsFromSignalMapping(txBlockingSignalsMap))
 	ctx = ctx.WithTxCompletionChannels(getChannelsFromSignalMapping(txCompletionSignalingMap))
+	ctx = ctx.WithTxBlockingChannels(getChannelsFromSignalMapping(txBlockingSignalsMap))
 
-	// Deliver the transaction and store the result in the channel
+	// ctx = ctx.WithTxMsgAccessOps(txMsgAccessOpMapping)
 
+	// Deliver the transaction and store the result in the channel
+	ctx.Logger().Info(fmt.Sprintf("Transactions Started=%d", txIndex))
 	resultChan <- ChannelResult{txIndex, app.DeliverTxWithResult(ctx, txBytes)}
 	metrics.IncrTxProcessTypeCounter(metrics.CONCURRENT)
+	ctx.Logger().Info(fmt.Sprintf("Transactions Finished=%d", txIndex))
 }
 
 func (app *App) ProcessBlockConcurrent(
 	ctx sdk.Context,
 	txs [][]byte,
+	// dependencyDag *acltypes.Dag,
 	completionSignalingMap map[int]acltypes.MessageCompletionSignalMapping,
 	blockingSignalsMap map[int]acltypes.MessageCompletionSignalMapping,
-) []*abci.ExecTxResult {
+) ([]*abci.ExecTxResult, bool) {
 	defer metrics.BlockProcessLatency(time.Now(), metrics.CONCURRENT)
 
 	var waitGroup sync.WaitGroup
@@ -1022,7 +1027,7 @@ func (app *App) ProcessBlockConcurrent(
 
 	// If there's no transactions then return empty results
 	if len(txs) == 0 {
-		return txResults
+		return txResults, true
 	}
 
 	// For each transaction, start goroutine and deliver TX
@@ -1036,6 +1041,9 @@ func (app *App) ProcessBlockConcurrent(
 			resultChan,
 			completionSignalingMap[txIndex],
 			blockingSignalsMap[txIndex],
+			// dependencyDag.CompletionSignalingMap[txIndex],
+			// dependencyDag.BlockingSignalsMap[txIndex],
+			// dependencyDag.TxMsgAccessOpMapping[txIndex],
 		)
 	}
 
@@ -1050,6 +1058,8 @@ func (app *App) ProcessBlockConcurrent(
 	// Gather Results and store it based on txIndex and read results from channel
 	// Concurrent results may be in different order than the original txIndex
 	txResultsMap := map[int]*abci.ExecTxResult{}
+
+	ctx.Logger().Info(fmt.Sprintf("Waiting for Transactions=%d", len(txs)))
 	for result := range resultChan {
 		txResultsMap[result.txIndex] = result.result
 	}
@@ -1059,10 +1069,19 @@ func (app *App) ProcessBlockConcurrent(
 		txResults = append(txResults, txResultsMap[txIndex])
 	}
 
-	return txResults
+	ok := true
+	for _, result := range txResults {
+		if result.GetCode() == sdkerrors.ErrInvalidConcurrencyExecution.ABCICode() {
+			ctx.Logger().Error(fmt.Sprintf("Invalid concurrent execution of deliverTx: %s", result.GetLog()))
+			ok = false
+		}
+	}
+
+	return txResults, ok
 }
 
 func (app *App) ProcessBlock(ctx sdk.Context, txs [][]byte, req BlockProcessRequest, lastCommit abci.CommitInfo) ([]abci.Event, []*abci.ExecTxResult, abci.ResponseEndBlock, error) {
+	// TODO:: Add concurrency validation for DEX MEM states
 	goCtx := app.decorateContextWithDexMemState(ctx.Context())
 	ctx = ctx.WithContext(goCtx).WithContextMemCache(sdk.NewContextMemCache())
 
@@ -1115,9 +1134,18 @@ func (app *App) ProcessBlock(ctx sdk.Context, txs [][]byte, req BlockProcessRequ
 		// CacheMultiStore where it writes the data to the parent store (DeliverState) in sorted Key order to maintain
 		// deterministic ordering between validators in the case of concurrent deliverTXs
 		processBlockCtx, processBlockCache := app.CacheContext(ctx)
-		txResults = app.ProcessBlockConcurrent(processBlockCtx, txs, dependencyDag.CompletionSignalingMap, dependencyDag.BlockingSignalsMap)
-		// Write the results back to the concurrent contexts
-		processBlockCache.Write()
+		// concurrentResults, ok := app.ProcessBlockConcurrent(processBlockCtx, txs, dependencyDag)
+		concurrentResults, ok := app.ProcessBlockConcurrent(processBlockCtx, txs, dependencyDag.CompletionSignalingMap, dependencyDag.BlockingSignalsMap)
+
+		if ok {
+			txResults = concurrentResults
+			// Write the results back to the concurrent contexts - if concurrent execution fails,
+			// this should not be called and the state is rolled back
+			processBlockCache.Write()
+		} else {
+			ctx.Logger().Info("Concurrent Execution failed, retrying with Synchronous")
+			txResults = app.ProcessBlockSynchronous(ctx, txs)
+		}
 	case acltypes.ErrGovMsgInBlock:
 		ctx.Logger().Info(fmt.Sprintf("Gov msg found while building DAG, processing synchronously: %s", err))
 		txResults = app.ProcessBlockSynchronous(ctx, txs)

go.mod

@@ -83,11 +83,13 @@ require (
 	github.com/improbable-eng/grpc-web v0.14.1 // indirect
 	github.com/inconshreveable/mousetrap v1.0.0 // indirect
 	github.com/jmhodges/levigo v1.0.0 // indirect
+	github.com/k0kubun/pp/v3 v3.2.0 // indirect
 	github.com/keybase/go-keychain v0.0.0-20190712205309-48d3d31d256d // indirect
 	github.com/klauspost/compress v1.15.1 // indirect
 	github.com/lib/pq v1.10.6 // indirect
 	github.com/libp2p/go-buffer-pool v0.0.2 // indirect
 	github.com/magiconair/properties v1.8.6 // indirect
+	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.16 // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369 // indirect
 	github.com/mitchellh/go-homedir v1.1.0 // indirect
@@ -131,7 +133,7 @@ require (
 )
 
 replace (
-	github.com/cosmos/cosmos-sdk => github.com/sei-protocol/sei-cosmos v0.1.200
+	github.com/cosmos/cosmos-sdk => github.com/sei-protocol/sei-cosmos v0.1.225
 	github.com/gogo/protobuf => github.com/regen-network/protobuf v1.3.3-alpha.regen.1
 	github.com/keybase/go-keychain => github.com/99designs/go-keychain v0.0.0-20191008050251-8e49817e8af4
 	github.com/tendermint/tendermint => github.com/sei-protocol/sei-tendermint v0.1.59