feat: removal message

proto/connect/marketmap/v2/tx.proto

@@ -33,6 +33,11 @@ service Msg {
   // Specifically if a market does not exist it will be created, otherwise it
   // will be updated. The response will be a map between ticker -> updated.
   rpc UpsertMarkets(MsgUpsertMarkets) returns (MsgUpsertMarketsResponse);
+
+  // RemoveMarkets removes the given markets from the marketmap if:
+  // - they exist in the map
+  // - they are disabled
+  rpc RemoveMarkets(MsgRemoveMarkets) returns (MsgRemoveMarketsResponse);
 }
 
 // MsgUpsertMarkets defines a message carrying a payload for performing market
@@ -134,3 +139,23 @@ message MsgRemoveMarketAuthorities {
 // MsgRemoveMarketAuthoritiesResponse defines the
 // Msg/RemoveMarketAuthoritiesResponse response type.
 message MsgRemoveMarketAuthoritiesResponse {}
+
+// MsgRemoveMarkets defines the Msg/RemoveMarkets request type. It contains the
+// new markets to be removed from the market map.
+message MsgRemoveMarkets {
+  option (cosmos.msg.v1.signer) = "admin";
+
+  // Admin defines the authority that is the x/marketmap
+  // Admin account.  This account is set in the module parameters.
+  string admin = 1 [ (cosmos_proto.scalar) = "cosmos.AddressString" ];
+
+  // Markets is the list of markets to remove.
+  repeated string markets = 2;
+}
+
+// MsgRemoveMarketsResponse defines the
+// Msg/MsgRemoveMarketsResponse response type.
+message MsgRemoveMarketsResponse {
+  // DeletedMarkets is the list of markets that were removed.
+  repeated string deleted_markets = 1;
+}

tests/integration/connect_ccv_suite.go

@@ -44,7 +44,7 @@ func NewConnectCCVIntegrationSuite(
 func (s *ConnectCCVSuite) TestCCVAggregation() {
 	ethusdc := connecttypes.NewCurrencyPair("ETH", "USDC")
 
-	s.Require().NoError(s.AddCurrencyPairs(s.chain, s.user, 3600, ethusdc))
+	s.Require().NoError(s.AddCurrencyPairs(s.chain, s.user, 3600, enabledTicker(ethusdc)))
 
 	cc, closeFn, err := GetChainGRPC(s.chain)
 	s.Require().NoError(err)

tests/integration/connect_setup.go

@@ -305,6 +305,33 @@ func QueryCurrencyPair(chain *cosmos.CosmosChain, cp connecttypes.CurrencyPair,
 	return res.Price, int64(res.Nonce), nil
 }
 
+// QueryMarket queries a market from the market map.
+func QueryMarket(chain *cosmos.CosmosChain, cp connecttypes.CurrencyPair) (mmtypes.Market, error) {
+	grpcAddr := chain.GetHostGRPCAddress()
+
+	// create the client
+	cc, err := grpc.Dial(grpcAddr, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		return mmtypes.Market{}, err
+	}
+	defer cc.Close()
+
+	// create the mm client
+	client := mmtypes.NewQueryClient(cc)
+
+	ctx := context.Background()
+
+	// query the currency pairs
+	res, err := client.Market(ctx, &mmtypes.MarketRequest{
+		CurrencyPair: cp,
+	})
+	if err != nil {
+		return mmtypes.Market{}, err
+	}
+
+	return res.Market, nil
+}
+
 // SubmitProposal creates and submits a proposal to the chain
 func SubmitProposal(chain *cosmos.CosmosChain, deposit sdk.Coin, submitter string, msgs ...sdk.Msg) (string, error) {
 	// build the proposal
@@ -359,30 +386,26 @@ func PassProposal(chain *cosmos.CosmosChain, propId string, timeout time.Duratio
 
 // AddCurrencyPairs creates + submits the proposal to add the given currency-pairs to state, votes for the prop w/ all nodes,
 // and waits for the proposal to pass.
-func (s *ConnectIntegrationSuite) AddCurrencyPairs(chain *cosmos.CosmosChain, user cosmos.User, price float64, cps ...connecttypes.CurrencyPair) error {
-	creates := make([]mmtypes.Market, len(cps))
-	for i, cp := range cps {
+func (s *ConnectIntegrationSuite) AddCurrencyPairs(chain *cosmos.CosmosChain, user cosmos.User, price float64,
+	tickers ...mmtypes.Ticker,
+) error {
+	creates := make([]mmtypes.Market, len(tickers))
+	for i, ticker := range tickers {
 		creates[i] = mmtypes.Market{
-			Ticker: mmtypes.Ticker{
-				CurrencyPair:     cp,
-				Decimals:         8,
-				MinProviderCount: 1,
-				Metadata_JSON:    "",
-				Enabled:          true,
-			},
+			Ticker: ticker,
 			ProviderConfigs: []mmtypes.ProviderConfig{
 				{
 					Name:           static.Name,
-					OffChainTicker: cp.String(),
+					OffChainTicker: ticker.String(),
 					Metadata_JSON:  fmt.Sprintf(`{"price": %f}`, price),
 				},
 			},
 		}
 	}
 
-	msg := &mmtypes.MsgCreateMarkets{
-		Authority:     s.user.FormattedAddress(),
-		CreateMarkets: creates,
+	msg := &mmtypes.MsgUpsertMarkets{
+		Authority: s.user.FormattedAddress(),
+		Markets:   creates,
 	}
 
 	tx := CreateTx(s.T(), s.chain, user, gasPrice, msg)
@@ -403,6 +426,37 @@ func (s *ConnectIntegrationSuite) AddCurrencyPairs(chain *cosmos.CosmosChain, us
 	return nil
 }
 
+func (s *ConnectIntegrationSuite) RemoveMarket(
+	chain *cosmos.CosmosChain,
+	markets []connecttypes.CurrencyPair,
+) error {
+	marketString := make([]string, len(markets))
+	for i, market := range markets {
+		marketString[i] = market.String()
+	}
+
+	msg := &mmtypes.MsgRemoveMarkets{
+		Admin:   s.user.FormattedAddress(),
+		Markets: marketString,
+	}
+
+	tx := CreateTx(s.T(), s.chain, s.user, gasPrice, msg)
+
+	// get an rpc endpoint for the chain
+	client := chain.Nodes()[0].Client
+	// broadcast the tx
+	resp, err := client.BroadcastTxCommit(context.Background(), tx)
+	if err != nil {
+		return err
+	}
+
+	if resp.TxResult.Code != abcitypes.CodeTypeOK {
+		return fmt.Errorf(resp.TxResult.Log)
+	}
+
+	return nil
+}
+
 func (s *ConnectIntegrationSuite) UpdateCurrencyPair(chain *cosmos.CosmosChain, markets []mmtypes.Market) error {
 	msg := &mmtypes.MsgUpsertMarkets{
 		Authority: s.user.FormattedAddress(),
@@ -451,7 +505,7 @@ func QueryProposal(chain *cosmos.CosmosChain, propID string) (*govtypesv1.QueryP
 	})
 }
 
-// WaitForProposalStatus, waits for the deposit period for the proposal to end
+// WaitForProposalStatus waits for the deposit period for the proposal to end
 func WaitForProposalStatus(chain *cosmos.CosmosChain, propID string, timeout time.Duration, status govtypesv1.ProposalStatus) error {
 	return testutil.WaitForCondition(timeout, 1*time.Second, func() (bool, error) {
 		prop, err := QueryProposal(chain, propID)