Problem: ibc relayer don't work well with default tx prioritization s…

…trategy (#848)

* Problem: ibc relayer don't work well with default tx prioritization strategy

Solution:
- Change the default priority mechanism to be based on gas price

* add priority integration test

* Update app/validator_tx_fee.go

* fix integration test

* fix integration test

* Update app/validator_tx_fee.go

CHANGELOG.md

@@ -10,6 +10,7 @@
 
 - Upgrade `cosmos-sdk` to `v0.45.6` and `ibc-go` to `v4.0.0-rc0`. #803
 - Changed go version to `1.18`. #803
+- Change the default priority mechanism to be based on gas price
 
 ### Security
 

app/app.go

@@ -643,6 +643,7 @@ func New(
 				FeegrantKeeper:  app.FeeGrantKeeper,
 				SignModeHandler: encodingConfig.TxConfig.SignModeHandler(),
 				SigGasConsumer:  ante.DefaultSigVerificationGasConsumer,
+				TxFeeChecker:    checkTxFeeWithValidatorMinGasPrices,
 			},
 			IBCKeeper: app.IBCKeeper,
 		},

app/validator_tx_fee.go

@@ -0,0 +1,62 @@
+package app
+
+import (
+	"math"
+
+	sdk "github.com/cosmos/cosmos-sdk/types"
+	sdkerrors "github.com/cosmos/cosmos-sdk/types/errors"
+)
+
+// checkTxFeeWithValidatorMinGasPrices implements the default fee logic, where the minimum price per
+// unit of gas is fixed and set by each validator, and the tx priority is computed from the gas price.
+func checkTxFeeWithValidatorMinGasPrices(ctx sdk.Context, tx sdk.Tx) (sdk.Coins, int64, error) {
+	feeTx, ok := tx.(sdk.FeeTx)
+	if !ok {
+		return nil, 0, sdkerrors.Wrap(sdkerrors.ErrTxDecode, "Tx must be a FeeTx")
+	}
+
+	feeCoins := feeTx.GetFee()
+	gas := feeTx.GetGas()
+
+	// Ensure that the provided fees meet a minimum threshold for the validator,
+	// if this is a CheckTx. This is only for local mempool purposes, and thus
+	// is only ran on check tx.
+	if ctx.IsCheckTx() {
+		minGasPrices := ctx.MinGasPrices()
+		if !minGasPrices.IsZero() {
+			requiredFees := make(sdk.Coins, len(minGasPrices))
+
+			// Determine the required fees by multiplying each required minimum gas
+			// price by the gas limit, where fee = ceil(minGasPrice * gasLimit).
+			for i, gp := range minGasPrices {
+				fee := gp.Amount.MulInt64(int64(gas))
+				requiredFees[i] = sdk.NewCoin(gp.Denom, fee.Ceil().RoundInt())
+			}
+
+			if !feeCoins.IsAnyGTE(requiredFees) {
+				return nil, 0, sdkerrors.Wrapf(sdkerrors.ErrInsufficientFee, "insufficient fees; got: %s required: %s", feeCoins, requiredFees)
+			}
+		}
+	}
+
+	priority := getTxPriority(feeCoins, int64(gas))
+	return feeCoins, priority, nil
+}
+
+// getTxPriority returns a naive tx priority based on the amount of the smallest denomination of the gas price
+// provided in a transaction.
+func getTxPriority(fee sdk.Coins, gas int64) int64 {
+	var priority int64
+	for _, c := range fee {
+		p := int64(math.MaxInt64)
+		gasPrice := c.Amount.QuoRaw(gas)
+		if gasPrice.IsInt64() {
+			p = gasPrice.Int64()
+		}
+		if priority == 0 || p < priority {
+			priority = p
+		}
+	}
+
+	return priority
+}

integration_tests/configs/mempool.yaml

@@ -0,0 +1,46 @@
+mempool-test:
+  config:
+    mempool:
+      version: v1
+    consensus:
+      timeout_commit: 5s
+  validators:
+    - coins: 10cro
+      staked: 10cro
+      commission_rate: "0.000000000000000000"
+    - coins: 10cro
+      staked: 10cro
+    # - coins: 1cro
+    #   staked: 1cro
+    #   min_self_delegation: 10000000  # 0.1cro
+  accounts:
+    - name: community
+      coins: 100cro
+    - name: ecosystem
+      coins: 200cro
+    - name: reserve
+      coins: 200cro
+      vesting: "60s"
+    - name: launch
+      coins: 100cro
+    - name: signer1
+      coins: 10000cro
+    - name: signer2
+      coins: 2000cro
+    - name: msigner1
+      coins: 2000cro
+    - name: msigner2
+      coins: 2000cro
+  genesis:
+    app_state:
+      staking:
+        params:
+          unbonding_time: "10s"
+      gov:
+        voting_params:
+          voting_period: "10s"
+        deposit_params:
+          max_deposit_period: "10s"
+          min_deposit:
+            - denom: "basecro"
+              amount: "10000000"

integration_tests/cosmoscli.py

@@ -1,6 +1,7 @@
 import json
 import tempfile
 
+import requests
 from pystarport import cluster, cosmoscli
 
 
@@ -72,6 +73,78 @@ def gov_propose_legacy(self, proposer, kind, proposal, no_validate=False, **kwar
                     )
                 )
 
+    def transfer(self, from_, to, coins, generate_only=False, **kwargs):
+        default_kwargs = {
+            "home": self.data_dir,
+            "keyring_backend": "test",
+            "chain_id": self.chain_id,
+            "node": self.node_rpc,
+        }
+        return json.loads(
+            self.raw(
+                "tx",
+                "bank",
+                "send",
+                from_,
+                to,
+                coins,
+                "-y",
+                "--generate-only" if generate_only else None,
+                **(default_kwargs | kwargs),
+            )
+        )
+
+    def sign_tx(self, tx_file, signer):
+        return json.loads(
+            self.raw(
+                "tx",
+                "sign",
+                tx_file,
+                from_=signer,
+                home=self.data_dir,
+                keyring_backend="test",
+                chain_id=self.chain_id,
+                node=self.node_rpc,
+            )
+        )
+
+    def sign_tx_json(self, tx, signer, max_priority_price=None):
+        if max_priority_price is not None:
+            tx["body"]["extension_options"].append(
+                {
+                    "@type": "/ethermint.types.v1.ExtensionOptionDynamicFeeTx",
+                    "max_priority_price": str(max_priority_price),
+                }
+            )
+        with tempfile.NamedTemporaryFile("w") as fp:
+            json.dump(tx, fp)
+            fp.flush()
+            return self.sign_tx(fp.name, signer)
+
+    def broadcast_tx(self, tx_file, **kwargs):
+        kwargs.setdefault("broadcast_mode", "block")
+        kwargs.setdefault("output", "json")
+        return json.loads(
+            self.raw("tx", "broadcast", tx_file, node=self.node_rpc, **kwargs)
+        )
+
+    def broadcast_tx_json(self, tx, **kwargs):
+        with tempfile.NamedTemporaryFile("w") as fp:
+            json.dump(tx, fp)
+            fp.flush()
+            return self.broadcast_tx(fp.name, **kwargs)
+
+    def tx_search_rpc(self, events: str):
+        node_rpc_http = "http" + self.node_rpc.removeprefix("tcp")
+        rsp = requests.get(
+            f"{node_rpc_http}/tx_search",
+            params={
+                "query": f'"{events}"',
+            },
+        ).json()
+        assert "error" not in rsp, rsp["error"]
+        return rsp["result"]["txs"]
+
 
 class ClusterCLI(cluster.ClusterCLI):
     def __init__(self, *args, **kwargs):
@@ -90,3 +163,6 @@ def cosmos_cli(self, i=0):
 
     def gov_propose_legacy(self, proposer, kind, proposal, i=0, **kwargs):
         return self.cosmos_cli(i).gov_propose_legacy(proposer, kind, proposal, **kwargs)
+
+    def transfer(self, from_, to, coins, i=0, generate_only=False, **kwargs):
+        return self.cosmos_cli(i).transfer(from_, to, coins, generate_only, **kwargs)

integration_tests/test_priority.py

@@ -0,0 +1,87 @@
+from pathlib import Path
+
+import pytest
+
+from .cosmoscli import ClusterCLI
+from .utils import cluster_fixture, wait_for_new_blocks
+
+PRIORITY_REDUCTION = 1000000
+
+
+pytestmark = pytest.mark.normal
+
+
+@pytest.fixture(scope="module")
+def cluster(worker_index, pytestconfig, tmp_path_factory):
+    "override cluster fixture for this test module"
+    yield from cluster_fixture(
+        Path(__file__).parent / "configs/mempool.yaml",
+        worker_index,
+        tmp_path_factory.mktemp("data"),
+    )
+
+
+def test_priority(cluster: ClusterCLI):
+    """
+    Check that prioritized mempool works, and the priority is decided by gas price.
+    """
+    cli = cluster.cosmos_cli()
+    test_cases = [
+        {
+            "from": cli.address("community"),
+            "to": cli.address("validator"),
+            "amount": "1000aphoton",
+            "gas_prices": "10basecro",
+            # if the priority is decided by fee, this tx will have the highest priority,
+            # if by gas price, it's the lowest.
+            "gas": 200000 * 10,
+        },
+        {
+            "from": cli.address("signer1"),
+            "to": cli.address("signer2"),
+            "amount": "1000aphoton",
+            "gas_prices": "20basecro",
+            "gas": 200000,
+        },
+        {
+            "from": cli.address("signer2"),
+            "to": cli.address("signer1"),
+            "amount": "1000aphoton",
+            "gas_prices": "30basecro",
+            "gas": 200000,
+        },
+    ]
+    txs = []
+    for tc in test_cases:
+        tx = cli.transfer(
+            tc["from"],
+            tc["to"],
+            tc["amount"],
+            gas_prices=tc["gas_prices"],
+            generate_only=True,
+            gas=tc["gas"],
+        )
+        txs.append(
+            cli.sign_tx_json(
+                tx, tc["from"], max_priority_price=tc.get("max_priority_price")
+            )
+        )
+
+    # wait for the beginning of a new block, so the window of time is biggest
+    # before the next block get proposed.
+    wait_for_new_blocks(cli, 1)
+
+    txhashes = []
+    for tx in txs:
+        rsp = cli.broadcast_tx_json(tx, broadcast_mode="sync")
+        assert rsp["code"] == 0, rsp["raw_log"]
+        txhashes.append(rsp["txhash"])
+
+    print("wait for two new blocks, so the sent txs are all included")
+    wait_for_new_blocks(cli, 2)
+
+    tx_results = [cli.tx_search_rpc(f"tx.hash='{txhash}'")[0] for txhash in txhashes]
+    tx_indexes = [(int(r["height"]), r["index"]) for r in tx_results]
+    print(tx_indexes)
+    # the first sent tx are included later, because of reversed priority order
+    assert all(i1 > i2 for i1, i2 in zip(tx_indexes, tx_indexes[1:]))