feat: support reject bucket migration (#493)

* feat: support reject bucket migration

* add test

* lint

* register MsgRejectMigrateBucket Gas param for PampasUpgrade

app/upgrade.go

@@ -4,6 +4,7 @@ import (
 	serverconfig "github.com/cosmos/cosmos-sdk/server/config"
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	"github.com/cosmos/cosmos-sdk/types/module"
+	"github.com/cosmos/cosmos-sdk/x/gashub/types"
 	upgradetypes "github.com/cosmos/cosmos-sdk/x/upgrade/types"
 
 	bridgemoduletypes "github.com/bnb-chain/greenfield/x/bridge/types"
@@ -79,6 +80,9 @@ func (app *App) registerPampasUpgradeHandler() {
 			app.CrossChainKeeper.SetChannelSendPermission(ctx, sdk.ChainID(app.appConfig.CrossChain.DestOpChainId), storagemoduletypes.ObjectChannelId, sdk.ChannelAllow)
 			app.CrossChainKeeper.SetChannelSendPermission(ctx, sdk.ChainID(app.appConfig.CrossChain.DestOpChainId), storagemoduletypes.GroupChannelId, sdk.ChannelAllow)
 
+			// register MsgRejectMigrateBucket Gas param
+			app.GashubKeeper.SetMsgGasParams(ctx, *types.NewMsgGasParamsWithFixedGas("/greenfield.storage.MsgRejectMigrateBucket", 1.2e3))
+
 			return app.mm.RunMigrations(ctx, app.configurator, fromVM)
 		})
 

e2e/tests/storage_test.go

@@ -2010,3 +2010,67 @@ func (s *StorageTestSuite) TestMaintenanceSPCreateBucketAndObject() {
 	s.Require().NoError(err)
 	s.Require().Equal(sptypes.STATUS_IN_SERVICE, spResp.StorageProvider.Status)
 }
+
+func (s *StorageTestSuite) TestRejectMigrateBucket() {
+	// construct bucket and object
+	primarySP := s.BaseSuite.PickStorageProvider()
+	gvg, found := primarySP.GetFirstGlobalVirtualGroup()
+	s.Require().True(found)
+	user := s.GenAndChargeAccounts(1, 1000000)[0]
+	bucketName := storageutils.GenRandomBucketName()
+	objectName := storageutils.GenRandomObjectName()
+	s.BaseSuite.CreateObject(user, primarySP, gvg.Id, bucketName, objectName)
+
+	var err error
+	dstPrimarySP := s.CreateNewStorageProvider()
+
+	// migrate bucket
+	msgMigrationBucket := storagetypes.NewMsgMigrateBucket(user.GetAddr(), bucketName, dstPrimarySP.Info.Id)
+	msgMigrationBucket.DstPrimarySpApproval.ExpiredHeight = math.MaxInt
+	msgMigrationBucket.DstPrimarySpApproval.Sig, err = dstPrimarySP.ApprovalKey.Sign(msgMigrationBucket.GetApprovalBytes())
+	s.SendTxBlock(user, msgMigrationBucket)
+	s.Require().NoError(err)
+
+	ctx := context.Background()
+	queryHeadBucketRequest := storagetypes.QueryHeadBucketRequest{
+		BucketName: bucketName,
+	}
+	queryHeadBucketResponse, err := s.Client.HeadBucket(ctx, &queryHeadBucketRequest)
+	s.Require().NoError(err)
+	s.Require().Equal(queryHeadBucketResponse.BucketInfo.BucketName, bucketName)
+	s.Require().Equal(queryHeadBucketResponse.BucketInfo.BucketStatus, storagetypes.BUCKET_STATUS_MIGRATING)
+
+	// Dest SP reject the migration
+	rejectMigration := storagetypes.NewMsgRejectMigrateBucket(dstPrimarySP.OperatorKey.GetAddr(), bucketName)
+	s.SendTxBlock(dstPrimarySP.OperatorKey, rejectMigration)
+	s.Require().NoError(err)
+
+	queryHeadBucketRequest = storagetypes.QueryHeadBucketRequest{
+		BucketName: bucketName,
+	}
+	queryHeadBucketResponse, err = s.Client.HeadBucket(ctx, &queryHeadBucketRequest)
+	s.Require().NoError(err)
+	s.Require().Equal(queryHeadBucketResponse.BucketInfo.BucketStatus, storagetypes.BUCKET_STATUS_CREATED)
+
+	// migrate bucket again
+	msgMigrationBucket = storagetypes.NewMsgMigrateBucket(user.GetAddr(), bucketName, dstPrimarySP.Info.Id)
+	msgMigrationBucket.DstPrimarySpApproval.ExpiredHeight = math.MaxInt
+	msgMigrationBucket.DstPrimarySpApproval.Sig, err = dstPrimarySP.ApprovalKey.Sign(msgMigrationBucket.GetApprovalBytes())
+	s.SendTxBlock(user, msgMigrationBucket)
+	s.Require().NoError(err)
+
+	// cancel migration by user
+	msgCancelMigrationBucket := storagetypes.NewMsgCancelMigrateBucket(user.GetAddr(), bucketName)
+	s.SendTxBlock(user, msgCancelMigrationBucket)
+	s.Require().NoError(err)
+
+	queryHeadBucketResponse, err = s.Client.HeadBucket(ctx, &queryHeadBucketRequest)
+	s.Require().NoError(err)
+	s.Require().Equal(queryHeadBucketResponse.BucketInfo.BucketStatus, storagetypes.BUCKET_STATUS_CREATED)
+
+	// dest SP should fail to reject
+	s.Client.SetKeyManager(dstPrimarySP.OperatorKey)
+	_, err = s.Client.BroadcastTx(context.Background(), []sdk.Msg{rejectMigration}, nil)
+	s.Require().Error(err)
+	s.Require().Equal(queryHeadBucketResponse.BucketInfo.BucketStatus, storagetypes.BUCKET_STATUS_CREATED)
+}

proto/greenfield/storage/events.proto

@@ -105,7 +105,7 @@ message EventCreateObject {
   string bucket_name = 3;
   // object_name define the name of object
   string object_name = 4;
-  // bucket_id define an u256 id for object
+  // bucket_id define an u256 id for bucket
   string bucket_id = 6 [
     (cosmos_proto.scalar) = "cosmos.Uint",
     (gogoproto.customtype) = "Uint",
@@ -499,7 +499,7 @@ message EventMigrationBucket {
   string operator = 1 [(cosmos_proto.scalar) = "cosmos.AddressString"];
   // The name of the bucket to be migrated
   string bucket_name = 2;
-  // bucket_id define an u256 id for object
+  // bucket_id define an u256 id for bucket
   string bucket_id = 3 [
     (cosmos_proto.scalar) = "cosmos.Uint",
     (gogoproto.customtype) = "Uint",
@@ -515,7 +515,20 @@ message EventCancelMigrationBucket {
   string operator = 1 [(cosmos_proto.scalar) = "cosmos.AddressString"];
   // The name of the bucket to be migrated
   string bucket_name = 2;
-  // bucket_id define an u256 id for object
+  // bucket_id define an u256 id for bucket
+  string bucket_id = 3 [
+    (cosmos_proto.scalar) = "cosmos.Uint",
+    (gogoproto.customtype) = "Uint",
+    (gogoproto.nullable) = false
+  ];
+}
+
+message EventRejectMigrateBucket {
+  // The address of the operator that reject the bucket migration, must be the dest SP
+  string operator = 1 [(cosmos_proto.scalar) = "cosmos.AddressString"];
+  // The name of the bucket to be migrated
+  string bucket_name = 2;
+  // bucket_id define an u256 id for bucket
   string bucket_id = 3 [
     (cosmos_proto.scalar) = "cosmos.Uint",
     (gogoproto.customtype) = "Uint",
@@ -529,7 +542,7 @@ message EventCompleteMigrationBucket {
   string operator = 1 [(cosmos_proto.scalar) = "cosmos.AddressString"];
   // The name of the bucket to be migrated
   string bucket_name = 2;
-  // bucket_id define an u256 id for object
+  // bucket_id define an u256 id for bucket
   string bucket_id = 3 [
     (cosmos_proto.scalar) = "cosmos.Uint",
     (gogoproto.customtype) = "Uint",

proto/greenfield/storage/tx.proto

@@ -58,6 +58,7 @@ service Msg {
   rpc MigrateBucket(MsgMigrateBucket) returns (MsgMigrateBucketResponse);
   rpc CompleteMigrateBucket(MsgCompleteMigrateBucket) returns (MsgCompleteMigrateBucketResponse);
   rpc CancelMigrateBucket(MsgCancelMigrateBucket) returns (MsgCancelMigrateBucketResponse);
+  rpc RejectMigrateBucket(MsgRejectMigrateBucket) returns (MsgRejectMigrateBucketResponse);
 }
 message MsgCreateBucket {
   option (cosmos.msg.v1.signer) = "creator";
@@ -615,3 +616,15 @@ message MsgCancelMigrateBucket {
 }
 
 message MsgCancelMigrateBucketResponse {}
+
+message MsgRejectMigrateBucket {
+  option (cosmos.msg.v1.signer) = "operator";
+
+  // operator defines the account address of the msg operator.
+  // only the Dest SP can send this transaction to reject the bucket migration.
+  string operator = 1 [(cosmos_proto.scalar) = "cosmos.AddressString"];
+  // bucket_name defines the name of the bucket that need to be migrated
+  string bucket_name = 2;
+}
+
+message MsgRejectMigrateBucketResponse {}

sdk/client/gnfd_tm.go

@@ -56,7 +56,6 @@ func (c *GreenfieldClient) GetStatus(ctx context.Context) (*ctypes.ResultStatus,
 
 func (c *GreenfieldClient) BroadcastVote(ctx context.Context, vote votepool.Vote) error {
 	_, err := c.tendermintClient.BroadcastVote(ctx, vote)
-
 	return err
 }
 

x/storage/keeper/keeper.go

@@ -2075,6 +2075,40 @@ func (k Keeper) CancelBucketMigration(ctx sdk.Context, operator sdk.AccAddress,
 	return nil
 }
 
+func (k Keeper) RejectBucketMigration(ctx sdk.Context, operator sdk.AccAddress, bucketName string) error {
+	store := ctx.KVStore(k.storeKey)
+	bucketInfo, found := k.GetBucketInfo(ctx, bucketName)
+	if !found {
+		return types.ErrNoSuchBucket
+	}
+	if bucketInfo.BucketStatus != types.BUCKET_STATUS_MIGRATING {
+		return types.ErrInvalidBucketStatus.Wrapf("The bucket is not in migrating status")
+	}
+
+	migrationBucketInfo, found := k.GetMigrationBucketInfo(ctx, bucketInfo.Id)
+	if !found {
+		return types.ErrMigrationBucketFailed.Wrapf("reject bucket migration failed due to the migrate bucket info not found.")
+	}
+
+	sp := k.spKeeper.MustGetStorageProvider(ctx, migrationBucketInfo.DstSpId)
+	if !sdk.MustAccAddressFromHex(sp.OperatorAddress).Equals(operator) {
+		return types.ErrAccessDenied.Wrap("Only the dest SP can reject the bucket migration.")
+	}
+
+	bucketInfo.BucketStatus = types.BUCKET_STATUS_CREATED
+	k.SetBucketInfo(ctx, bucketInfo)
+	store.Delete(types.GetMigrationBucketKey(bucketInfo.Id))
+
+	if err := ctx.EventManager().EmitTypedEvents(&types.EventRejectMigrateBucket{
+		Operator:   operator.String(),
+		BucketName: bucketName,
+		BucketId:   bucketInfo.Id,
+	}); err != nil {
+		return err
+	}
+	return nil
+}
+
 func (k Keeper) GetMigrationBucketInfo(ctx sdk.Context, bucketID sdkmath.Uint) (*types.MigrationBucketInfo, bool) {
 	store := ctx.KVStore(k.storeKey)
 

x/storage/keeper/msg_server.go

@@ -688,6 +688,19 @@ func (k msgServer) CancelMigrateBucket(goCtx context.Context, msg *types.MsgCanc
 	return &types.MsgCancelMigrateBucketResponse{}, nil
 }
 
+func (k msgServer) RejectMigrateBucket(goCtx context.Context, msg *storagetypes.MsgRejectMigrateBucket) (*storagetypes.MsgRejectMigrateBucketResponse, error) {
+	ctx := sdk.UnwrapSDKContext(goCtx)
+
+	operator := sdk.MustAccAddressFromHex(msg.Operator)
+
+	err := k.RejectBucketMigration(ctx, operator, msg.BucketName)
+	if err != nil {
+		return nil, err
+	}
+
+	return &types.MsgRejectMigrateBucketResponse{}, nil
+}
+
 func (k Keeper) verifyGVGSignatures(ctx sdk.Context, bucketID math.Uint, dstSP *sptypes.StorageProvider, gvgMappings []*storagetypes.GVGMapping) error {
 	// verify secondary sp signature
 	for _, newLvg2gvg := range gvgMappings {

x/storage/types/codec.go

@@ -26,6 +26,7 @@ func RegisterCodec(cdc *codec.LegacyAmino) {
 	cdc.RegisterConcrete(&MsgMigrateBucket{}, "storage/MigrateBucket", nil)
 	cdc.RegisterConcrete(&MsgCompleteMigrateBucket{}, "storage/CompleteMigrateBucket", nil)
 	cdc.RegisterConcrete(&MsgCancelMigrateBucket{}, "storage/CancelMigrateBucket", nil)
+	cdc.RegisterConcrete(&MsgRejectMigrateBucket{}, "storage/RejectMigrateBucket", nil)
 	// this line is used by starport scaffolding # 2
 }
 
@@ -112,6 +113,9 @@ func RegisterInterfaces(registry cdctypes.InterfaceRegistry) {
 	registry.RegisterImplementations((*sdk.Msg)(nil),
 		&MsgCancelMigrateBucket{},
 	)
+	registry.RegisterImplementations((*sdk.Msg)(nil),
+		&MsgRejectMigrateBucket{},
+	)
 	// this line is used by starport scaffolding # 3
 
 	msgservice.RegisterMsgServiceDesc(registry, &_Msg_serviceDesc)